theater/cabletest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cable verdict folded into one console cell (length green, fault name red, xover amber, fail; diag rule lines and counters-reset line gone), reset fires at the press not diag completion, and the measure's blip is suppressed at every counter source instead of counted-hidden-reverted: worker error adds, loss-window write-offs (quiet slides state without charging; leaving the quiet era presumes in-window holes delivered so the gap's tail is never charged late), NIC poller tracks raw without accumulating, socket drops discarded, modules un-primed by the measure itself; display hold deleted so pre-existing errors no longer blink out during a measure

Browse Source
This commit is contained in:
flamingcow
2026-08-17 11:25:31 -07:00
parent 3d7105073e
commit e3c3f945c5
10 changed files with 220 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.go
+54 -52
View File
@@ -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))