diff --git a/counters.go b/counters.go index af3655b..ebb2866 100644 --- a/counters.go +++ b/counters.go @@ -105,13 +105,14 @@ func sumFields(base string, fields []string) uint64 { const nicInterval = sampleInterval type nicPoller struct { - txBase string - rx *statReader - total *atomic.Uint64 - raw uint64 + txBase string + rx *statReader + total *atomic.Uint64 + measuring *atomic.Bool + raw uint64 } -func newNICPoller(fd int, txName, rxName string, total *atomic.Uint64) (*nicPoller, error) { +func newNICPoller(fd int, txName, rxName string, total *atomic.Uint64, measuring *atomic.Bool) (*nicPoller, error) { drv, err := ifDriver(rxName) if err != nil { return nil, err @@ -125,9 +126,10 @@ func newNICPoller(fd int, txName, rxName string, total *atomic.Uint64) (*nicPoll return nil, err } return &nicPoller{ - txBase: "/sys/class/net/" + txName, - rx: rx, - total: total, + txBase: "/sys/class/net/" + txName, + rx: rx, + total: total, + measuring: measuring, }, nil } @@ -142,10 +144,12 @@ func (p *nicPoller) read() uint64 { // 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. +// 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 { + if raw > p.raw && !p.measuring.Load() { p.total.Add(raw - p.raw) } p.raw = raw diff --git a/docs/state.md b/docs/state.md index 7b08ce1..4ab206d 100644 --- a/docs/state.md +++ b/docs/state.md @@ -75,5 +75,5 @@ In `~/work/` alongside the phydiag artifacts, ready to fold into the repo's `ker ## Standing goals - **Length is a product goal again** — the FS (ECD length) rides alongside the Wiitek (SNR) in the test pair, which the tree implements. The Fibergaga (Aquantia 1E.C884 ±1 m + 1E.C800 verdicts, documented) remains the alternate length path if the pair mix ever changes. -- **BCM ECD works** (recipe recovered from the OpenBCM SDK, validated on the FS — modules/fs/): per-pair lengths meter-accurate against a known ~45 m cable. cabletest runs it at bringup and on every reset, re-baselining counters after the relink so the blip is never charged (`phy.go`). Remaining work is characterizing the link blip the run causes (length stays a between-measurements operation until then — open-questions). +- **BCM ECD works** (recipe recovered from the OpenBCM SDK, validated on the FS — modules/fs/): per-pair lengths meter-accurate against a known ~45 m cable. cabletest runs it at bringup and on every reset; while a measure runs, every failure counter is suppressed at its source (worker error adds, loss-window write-offs, NIC-poller accumulation, socket drops, module latch priming), so the blip is never counted anywhere rather than counted and reverted (`phy.go` `cableDiag.measuring`). Remaining work is characterizing the link blip the run causes (length stays a between-measurements operation until then — open-questions). - **Pre-FEC verification** on the Aquantia — counters documented; needs the graded-noise correlation run (design: modules/fibergaga/). diff --git a/loss.go b/loss.go index aec5dca..f675fb4 100644 --- a/loss.go +++ b/loss.go @@ -12,12 +12,13 @@ const ( ) type lossWindow struct { - mu sync.Mutex - base uint64 - inited bool - bits []uint64 - lost atomic.Uint64 - late atomic.Uint64 + mu sync.Mutex + base uint64 + inited bool + quietEra 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 @@ -47,11 +48,24 @@ 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. -func (w *lossWindow) observe(seq uint64) bool { +// 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 { if seq >= w.sent.Load() { return false } w.mu.Lock() + if w.quietEra && !quiet { + // 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. + for i := range w.bits { + w.bits[i] = ^uint64(0) + } + } + w.quietEra = quiet 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 @@ -63,11 +77,13 @@ func (w *lossWindow) observe(seq uint64) bool { } if seq < w.base { w.mu.Unlock() - w.late.Add(1) + if !quiet { + w.late.Add(1) + } return true } if seq >= w.base+lossSlots { - w.evict(seq - lossSlots + 1) + w.evict(seq-lossSlots+1, quiet) } idx := seq & (lossSlots - 1) w.bits[idx>>6] |= 1 << (idx & 63) @@ -75,7 +91,7 @@ func (w *lossWindow) observe(seq uint64) bool { return true } -func (w *lossWindow) evict(newBase uint64) { +func (w *lossWindow) evict(newBase uint64, quiet bool) { span := newBase - w.base if span >= lossSlots { var missing uint64 @@ -83,7 +99,9 @@ func (w *lossWindow) evict(newBase uint64) { missing += uint64(64 - bits.OnesCount64(w.bits[i])) w.bits[i] = 0 } - w.lost.Add(missing + span - lossSlots) + if !quiet { + w.lost.Add(missing + span - lossSlots) + } w.base = newBase return } @@ -97,6 +115,8 @@ func (w *lossWindow) evict(newBase uint64) { } w.bits[word] &^= bit } - w.lost.Add(missing) + if !quiet { + w.lost.Add(missing) + } w.base = newBase } diff --git a/loss_test.go b/loss_test.go index 5f8bd3e..12291f0 100644 --- a/loss_test.go +++ b/loss_test.go @@ -14,7 +14,7 @@ func newWindow(sent uint64) *lossWindow { func TestLossWindowContiguousLosesNothing(t *testing.T) { w := newWindow(lossSlots + 1000) for seq := uint64(0); seq < lossSlots+1000; seq++ { - w.observe(seq) + w.observe(seq, false) } if got := w.lost.Load(); got != 0 { t.Errorf("lost = %d, want 0", got) @@ -30,7 +30,7 @@ func TestLossWindowCountsGapOnceEvicted(t *testing.T) { if seq == 100 { continue } - w.observe(seq) + w.observe(seq, false) } if got := w.lost.Load(); got != 1 { t.Errorf("lost = %d, want 1", got) @@ -42,13 +42,13 @@ func TestLossWindowCountsGapOnceEvicted(t *testing.T) { func TestLossWindowOutOfOrderIsNotLoss(t *testing.T) { w := newWindow(lossSlots + 1000) for seq := uint64(99); ; seq-- { - w.observe(seq) + w.observe(seq, false) if seq == 0 { break } } for seq := uint64(100); seq < lossSlots+1000; seq++ { - w.observe(seq) + w.observe(seq, false) } if got := w.lost.Load(); got != 0 { t.Errorf("lost = %d, want 0", got) @@ -59,8 +59,8 @@ func TestLossWindowOutOfOrderIsNotLoss(t *testing.T) { // the window held plus the sequences that never landed in it at all. func TestLossWindowJumpBeyondWindow(t *testing.T) { w := newWindow(200001) - w.observe(0) - w.observe(200000) + w.observe(0, false) + w.observe(200000, false) // Everything below the new base except the one sequence that was seen. want := uint64(200000 - lossSlots + 1 - 1) @@ -74,8 +74,8 @@ func TestLossWindowJumpBeyondWindow(t *testing.T) { func TestLossWindowBelowBaseIsLate(t *testing.T) { w := newWindow(100001) - w.observe(100000) - w.observe(1000) + w.observe(100000, false) + w.observe(1000, false) if got := w.late.Load(); got != 1 { t.Errorf("late = %d, want 1", got) } @@ -88,7 +88,7 @@ func TestLossWindowBelowBaseIsLate(t *testing.T) { // another worker is still holding land inside rather than arriving late. func TestLossWindowStartsHalfAWindowBack(t *testing.T) { w := newWindow(100001) - w.observe(100000) + w.observe(100000, false) if w.base != 100000-lossSlots/2 { t.Errorf("base = %d, want %d", w.base, 100000-lossSlots/2) } @@ -103,11 +103,11 @@ func TestLossWindowRefusesUnsentSeq(t *testing.T) { const first = 100000 w := newWindow(first + 2000) for seq := uint64(first); seq < first+1000; seq++ { - w.observe(seq) + w.observe(seq, false) } base := w.base - if w.observe(1 << 62) { + if w.observe(1<<62, false) { t.Error("a sequence number far past the sender's frontier was accepted") } if got := w.lost.Load(); got != 0 { @@ -121,7 +121,7 @@ func TestLossWindowRefusesUnsentSeq(t *testing.T) { // Still tracking the real traffic, rather than reporting every frame late // against a base that ran away. for seq := uint64(first + 1000); seq < first+2000; seq++ { - w.observe(seq) + w.observe(seq, false) } if got := w.late.Load(); got != 0 { t.Errorf("late = %d, want 0", got) @@ -131,14 +131,42 @@ func TestLossWindowRefusesUnsentSeq(t *testing.T) { } } +// A gap written off while quiet — 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) { + w := newWindow(4 * lossSlots) + for seq := uint64(0); seq < 100; seq++ { + w.observe(seq, false) + } + // 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) + } + + // The measure ends immediately — the worst case, with the gap's tail + // still in-window — and a single real gap follows. + for seq := uint64(2*lossSlots + 1); seq < 4*lossSlots-100; seq++ { + if seq == 2*lossSlots+500 { + continue + } + w.observe(seq, false) + } + if got := w.lost.Load(); got != 1 { + t.Errorf("lost = %d, want exactly the one real post-measure gap", got) + } +} + // The sender's own frontier is the bound, so the sequence one past it is // refused while the one below it is not. func TestLossWindowBoundIsExclusive(t *testing.T) { w := newWindow(500) - if !w.observe(499) { + if !w.observe(499, false) { t.Error("the last sequence the sender put on the wire was refused") } - if w.observe(500) { + if w.observe(500, false) { t.Error("a sequence the sender had not reached was accepted") } } diff --git a/main.go b/main.go index 55a66c8..e5650ca 100644 --- a/main.go +++ b/main.go @@ -39,6 +39,10 @@ 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 + // 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. @@ -250,22 +254,26 @@ func (d *direction) reset() { } // Returns the new start time, so the uptime shown alongside the totals counts -// from the reset rather than from launch. -func resetAll(dirs []*direction, mods []*phyModule, stats *streamTable) time.Time { +// from the reset rather than from launch; the elapsed clock restarting is the +// visible mark of the re-baseline. +func resetAll(dirs []*direction, mods []*phyModule) time.Time { for _, d := range dirs { d.reset() } for _, m := range mods { m.reset() } - stats.sinceHeader = 0 - fmt.Println(stats.rule("counters reset")) return time.Now() } +// 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 { - d.drops += packetDrops(fd) + n := packetDrops(fd) + if !d.measuring.Load() { + d.drops += n + } } } @@ -286,7 +294,7 @@ var intervalCols = []colSpec{ {group: "OVERALL", title: "elapsed", width: 9, right: true}, {group: "OVERALL", title: "packets", width: 9, right: true}, {group: "OVERALL", title: "bytes", width: 9, right: true}, - {group: "OVERALL", title: "metres", width: 6, right: true}, + {group: "OVERALL", title: "cable", width: 6, right: true}, {group: "OVERALL", title: "corrected", width: 9, right: true}, {group: "OVERALL", title: "lost", width: 9, right: true}, {group: "OVERALL", title: "corrupt", width: 9, right: true}, @@ -331,17 +339,11 @@ func (d *direction) counters(now counterSet) view { } } -// Errors seen while a measure is in flight are the diag's own link blip and -// are re-based away at its completion; until then they are held off the -// display rather than shown as the cable's. -func measureView(diag *cableDiag, modules []*phyModule, v view) (view, phyDisplay) { +// Nothing to hold or revert here: failures during a measure were never +// counted, so the view is always the counters as they stand. +func phyView(diag *cableDiag, modules []*phyModule) phyDisplay { info, measuring := diag.snapshot() - phy := phyDisplayFrom(info, measuring, modules[0].view(), modules[1].view()) - if measuring { - v.window, v.since = errs{}, errs{} - phy.corrected, phy.recent = 0, 0 - } - return v, phy + return phyDisplayFrom(info, measuring, modules[0].view(), modules[1].view()) } func totalView(views []view) view { @@ -417,7 +419,7 @@ func (d *direction) primeCounters() { d.reset() } -func buildDirection(label string, tx, rx endpoint) (*direction, error) { +func buildDirection(label string, tx, rx endpoint, measuring *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) @@ -425,8 +427,9 @@ func buildDirection(label string, tx, rx endpoint) (*direction, error) { txs[i] = &txStats{} } d := &direction{ - txStats: txs, - streams: newLossWindows(txs), + txStats: txs, + streams: newLossWindows(txs), + measuring: measuring, } // 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. @@ -436,7 +439,7 @@ func buildDirection(label string, tx, rx endpoint) (*direction, error) { } d.statFD = statFD - d.poller, err = newNICPoller(statFD, tx.name, rx.name, &d.nic) + d.poller, err = newNICPoller(statFD, tx.name, rx.name, &d.nic, measuring) if err != nil { return nil, fmt.Errorf("%s: %w", label, err) } @@ -466,12 +469,13 @@ func buildDirection(label string, tx, rx endpoint) (*direction, error) { 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], - startTx: startTx, + fd: fd, + stream: uint16(i), + spec: d.specs[i], + batch: batchSize, + stats: d.txStats[i], + measuring: d.measuring, + startTx: startTx, } wg.Add(1) go func() { @@ -482,13 +486,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], - ready: rxReady, + fd: fd, + batch: batchSize, + stream: uint16(i), + spec: d.specs[i], + stats: d.rxStats[i], + loss: &d.streams[i], + measuring: d.measuring, + ready: rxReady, } wg.Add(1) go func() { @@ -657,7 +662,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]) + d, err := buildDirection(p[0].name+"->"+p[1].name, p[0], p[1], &diag.measuring) if err != nil { return err } @@ -768,10 +773,18 @@ func run(aName, bName string) (err error) { } start := time.Now() + // 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. + kickMeasure := func() { + if diag.kick(&done) { + start = resetAll(dirs, modules) + } + } close(startTx) // The first measure rides the same async path as a reset, so startup never - // waits on it; counters re-baseline when its link blip is over. - diag.kick(&done) + // waits on it. + kickMeasure() tick := time.NewTicker(reportInterval) defer tick.Stop() @@ -784,32 +797,21 @@ func run(aName, bName string) (err error) { return fmt.Errorf("%v", p) case <-sig: return nil - // A reset re-measures the cable first; the counters re-baseline at diag - // completion, so its link blip is never charged to the fresh run. + // The verdict lives in the cable cell, not its own line. case <-space: - if diag.kick(&done) { - fmt.Println(stats.rule("measuring cable")) - } - case err := <-diag.completed: - if err != nil { - fmt.Println(stats.rule("cable diag failed: " + err.Error())) - } else { - info, _ := diag.snapshot() - fmt.Println(stats.rule("cable diag: " + cableLine(info))) - } - start = resetAll(dirs, modules, stats) + kickMeasure() case <-disp.fb.flips: now := time.Now() px, py, down := touch.get() x, y := disp.fb.fromPanel(px, py) if disp.holdReset(x, y, down, now) { - diag.kick(&done) + kickMeasure() } disp.showVersion = down && disp.versionSpot.contains(x, y) for i, d := range dirs { views[i] = d.displayView() } - v, phy := measureView(diag, modules, totalView(views)) + v, phy := totalView(views), phyView(diag, modules) if err := disp.render(v, now.Sub(start), phy, noise.view()); err != nil { return err @@ -819,7 +821,7 @@ func run(aName, bName string) (err error) { for i, d := range dirs { rows[i] = d.displayView() } - v, phy := measureView(diag, modules, totalView(rows)) + v, phy := totalView(rows), phyView(diag, modules) for _, m := range modules { for _, n := range m.takeNotes() { fmt.Println(stats.rule(n)) diff --git a/phy.go b/phy.go index 02b0239..9d7d4c1 100644 --- a/phy.go +++ b/phy.go @@ -769,6 +769,15 @@ func (m *phyModule) reset() { m.mu.Unlock() } +// Latches drained after a measure belong to its blip: un-priming makes the +// next poll an origin only, without discarding the pre-measure totals. +func (m *phyModule) forgive() { + m.mu.Lock() + m.primed = false + m.recentDelta = 0 + m.mu.Unlock() +} + type phyModView struct { fresh bool link bool @@ -802,6 +811,7 @@ type cableInfo struct { ecd ecdResult maps [2]byte haveMaps [2]bool + failed bool } func (c cableInfo) metresString() string { @@ -845,8 +855,6 @@ type phyDisplay struct { metresClass int } -func pairLetter(i int) string { return string(rune('A' + i)) } - // Each end resolves MDI on its own, so a swap at either known end counts; an // end with no readable map abstains. func pairSwapped(i int, c cableInfo) bool { @@ -858,32 +866,35 @@ func pairSwapped(i int, c cableInfo) bool { return false } +// The whole cable verdict in one cell: the length when healthy, the first +// faulted pair's verdict when not, "xover" for a pair swap, "fail" for a diag +// that never delivered. func cableSummary(cable cableInfo, measuring bool) (string, int) { if measuring { - return "...", clsNone + return "-", clsNone } - anyData, anyFault, anySwap := false, false, false + if cable.failed { + return "fail", clsBad + } + anyData, anySwap := false, false for i, v := range cable.ecd.verdicts { if v != 0 { anyData = true } if v != 0 && v != pairOK { - anyFault = true + return pairVerdicts[v], clsBad } if pairSwapped(i, cable) { anySwap = true } } - s := cable.metresString() switch { case !anyData: return "-", clsNone - case anyFault: - return s, clsBad case anySwap: - return s, clsWarn + return "xover", clsWarn } - return s, clsGood + return cable.metresString(), clsGood } // The margin is the worst pair across the ends that measure SNR (the Wiitek's @@ -977,6 +988,7 @@ func measureCable(mods []*phyModule, waitRelink bool, done *atomic.Bool) (cableI } defer func() { for _, m := range mods { + m.forgive() m.busy.Store(false) } }() @@ -1010,83 +1022,45 @@ func measureCable(mods []*phyModule, waitRelink bool, done *atomic.Bool) (cableI } type cableDiag struct { - mods []*phyModule - completed chan error + mods []*phyModule - mu sync.Mutex - info cableInfo - running bool + // 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 + + mu sync.Mutex + info cableInfo } func newCableDiag(mods []*phyModule, info cableInfo) *cableDiag { - return &cableDiag{mods: mods, completed: make(chan error, 1), info: info} + return &cableDiag{mods: mods, info: info} } func (c *cableDiag) snapshot() (cableInfo, bool) { c.mu.Lock() defer c.mu.Unlock() - return c.info, c.running + return c.info, c.measuring.Load() } func (c *cableDiag) kick(done *atomic.Bool) bool { - c.mu.Lock() - if c.running { - c.mu.Unlock() + if !c.measuring.CompareAndSwap(false, true) { return false } - c.running = true - c.mu.Unlock() - go func() { defer holdPanic() info, _, err := measureCable(c.mods, true, done) if err != nil { - info = cableInfo{} + info = cableInfo{failed: true} } c.mu.Lock() c.info = info - c.running = false c.mu.Unlock() - select { - case c.completed <- err: - default: - } + c.measuring.Store(false) }() return true } -func mapString(m byte, have bool) string { - if !have { - return "unread" - } - if m == pairIdentityMap { - return "straight" - } - out := make([]string, 4) - for i := range out { - out[i] = pairLetter(int(m>>(2*i)) & 3) - } - return "swapped to " + strings.Join(out, "") -} - -func verdictString(r ecdResult) string { - bad := []string{} - for i, v := range r.verdicts { - if v != pairOK { - s, ok := pairVerdicts[v] - if !ok { - s = fmt.Sprintf("%d", v) - } - bad = append(bad, fmt.Sprintf("%s %s at %dm", pairLetter(i), s, r.metres[i])) - } - } - if len(bad) > 0 { - return strings.Join(bad, ", ") - } - return fmt.Sprintf("all pairs ok, %d/%d/%d/%d m", - r.metres[0], r.metres[1], r.metres[2], r.metres[3]) -} - func openModules(names [2]string) ([]*phyModule, [2]string, error) { mods := make([]*phyModule, 0, 2) var idents [2]string @@ -1218,7 +1192,3 @@ func moduleChecks(mods []*phyModule, names [2]string) []checkResult { return out } -func cableLine(c cableInfo) string { - return fmt.Sprintf("%s; map %s / %s", - verdictString(c.ecd), mapString(c.maps[0], c.haveMaps[0]), mapString(c.maps[1], c.haveMaps[1])) -} diff --git a/phy_test.go b/phy_test.go index 21d63ca..586c9ea 100644 --- a/phy_test.go +++ b/phy_test.go @@ -19,8 +19,8 @@ func TestPhyDisplayWorstMarginAndFault(t *testing.T) { if !d.haveSNR || d.worstMargin != 0.5 { t.Errorf("worstMargin = %v (have %v), want 0.5", d.worstMargin, d.haveSNR) } - if d.metres != "44" || d.metresClass != clsBad { - t.Errorf("metres = %q class %d, want the healthy mean 44 painted bad", d.metres, d.metresClass) + if d.metres != "OPEN" || d.metresClass != clsBad { + t.Errorf("cable cell = %q class %d, want the fault named bad", d.metres, d.metresClass) } } @@ -51,10 +51,13 @@ func TestCableSummary(t *testing.T) { class int }{ {"clean", cableInfo{ecd: healthy, maps: [2]byte{pairIdentityMap, pairIdentityMap}, haveMaps: [2]bool{true, true}}, false, "49", clsGood}, - {"far-end swap", cableInfo{ecd: healthy, maps: [2]byte{pairIdentityMap, 0xE1}, haveMaps: [2]bool{true, true}}, false, "49", clsWarn}, + {"far-end swap", cableInfo{ecd: healthy, maps: [2]byte{pairIdentityMap, 0xE1}, haveMaps: [2]bool{true, true}}, false, "xover", clsWarn}, {"one-end map only", cableInfo{ecd: healthy, maps: [2]byte{pairIdentityMap, 0}, haveMaps: [2]bool{true, false}}, false, "49", clsGood}, + {"fault", cableInfo{ecd: ecdResult{verdicts: [4]int{pairOK, pairShort, pairXtalk, pairOK}, metres: [4]int{50, 12, 30, 51}}}, false, "SHORT", clsBad}, + {"fault beats swap", cableInfo{ecd: ecdResult{verdicts: [4]int{pairOpen, pairOK, pairOK, pairOK}}, maps: [2]byte{pairIdentityMap, 0xE1}, haveMaps: [2]bool{true, true}}, false, "OPEN", clsBad}, + {"diag failed", cableInfo{failed: true}, false, "fail", clsBad}, {"no diag yet", cableInfo{}, false, "-", clsNone}, - {"measuring", cableInfo{ecd: healthy, maps: [2]byte{pairIdentityMap, pairIdentityMap}, haveMaps: [2]bool{true, true}}, true, "...", clsNone}, + {"measuring", cableInfo{ecd: healthy, maps: [2]byte{pairIdentityMap, pairIdentityMap}, haveMaps: [2]bool{true, true}}, true, "-", clsNone}, } { s, cl := cableSummary(c.cable, c.measuring) if s != c.want || cl != c.class { @@ -104,13 +107,3 @@ func TestSNRClass(t *testing.T) { } } -func TestVerdictString(t *testing.T) { - ok := ecdResult{verdicts: [4]int{1, 1, 1, 1}, metres: [4]int{45, 45, 41, 46}} - if got := verdictString(ok); got != "all pairs ok, 45/45/41/46 m" { - t.Errorf("healthy = %q", got) - } - bad := ecdResult{verdicts: [4]int{1, 2, 4, 1}, metres: [4]int{45, 12, 30, 46}} - if got := verdictString(bad); got != "B OPEN at 12m, C XTALK at 30m" { - t.Errorf("faulted = %q", got) - } -} diff --git a/rx.go b/rx.go index b7ab65c..5d11a78 100644 --- a/rx.go +++ b/rx.go @@ -110,13 +110,14 @@ func fillBack(frames, bytes []uint64) { } type rxWorker struct { - fd int - batch int - stream uint16 - spec *frameSpec - stats *rxStats - loss *lossWindow - ready *sync.WaitGroup + fd int + batch int + stream uint16 + spec *frameSpec + stats *rxStats + loss *lossWindow + measuring *atomic.Bool + ready *sync.WaitGroup } func (w *rxWorker) run(done *atomic.Bool) { @@ -132,9 +133,12 @@ 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() n, err := recvmmsg(w.fd, hdrs, unix.MSG_WAITFORONE) if n <= 0 { - if err != nil && err != unix.EAGAIN && err != unix.EINTR { + if err != nil && err != unix.EAGAIN && err != unix.EINTR && !quiet { w.stats.rxErrs.Add(1) } continue @@ -145,6 +149,9 @@ 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 { @@ -158,18 +165,22 @@ 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) { - w.stats.badHdr.Add(1) + if p.stream != w.stream || !w.loss.observe(p.seq, quiet) { + if !quiet { + w.stats.badHdr.Add(1) + } continue } want, ok := w.spec.expectedCRC(p.patIdx, p.payLen) if !ok { - w.stats.badLen.Add(1) + if !quiet { + w.stats.badLen.Add(1) + } continue } pay := buf[minFrame : minFrame+p.payLen] - if crc32.Checksum(pay, crcTable) != want { + if crc32.Checksum(pay, crcTable) != want && !quiet { w.stats.crcErr.Add(1) } } diff --git a/tx.go b/tx.go index 59adc8f..a9670bc 100644 --- a/tx.go +++ b/tx.go @@ -17,12 +17,13 @@ type txStats struct { } type txWorker struct { - fd int - stream uint16 - spec *frameSpec - batch int - stats *txStats - startTx <-chan struct{} + fd int + stream uint16 + spec *frameSpec + batch int + stats *txStats + measuring *atomic.Bool + startTx <-chan struct{} } func (w *txWorker) run(done *atomic.Bool) { @@ -63,7 +64,8 @@ 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 { + if n < 0 && err != unix.EINTR && err != unix.EAGAIN && err != unix.ENOBUFS && + !w.measuring.Load() { w.stats.errs.Add(1) } } diff --git a/ui.go b/ui.go index 5c3d56a..382860c 100644 --- a/ui.go +++ b/ui.go @@ -386,10 +386,12 @@ func (d *display) errCounts(x, w, y int, e errs) int { func metresStat(phy phyDisplay) statCell { col := uiFg + unit := "m" if phy.metresClass != clsGood { col = classColor(phy.metresClass) + unit = "cable" } - return statCell{phy.metres, "m", col} + return statCell{phy.metres, unit, col} } func (d *display) panel(p rect, bad bool) (int, int) {