From 7c09c345640989e486ab88aefd8e19b2415a785a Mon Sep 17 00:00:00 2001 From: flamingcow Date: Sat, 25 Jul 2026 18:33:36 -0700 Subject: [PATCH] Stream forever with eviction-based per-stream loss detection --- loss.go | 81 +++++++++++++++++++++++++++++ main.go | 150 ++++++++++-------------------------------------------- render.go | 20 +++----- rx.go | 17 +------ 4 files changed, 117 insertions(+), 151 deletions(-) create mode 100644 loss.go diff --git a/loss.go b/loss.go new file mode 100644 index 0000000..41fda99 --- /dev/null +++ b/loss.go @@ -0,0 +1,81 @@ +package main + +import ( + "math/bits" + "sync" + "sync/atomic" +) + +const ( + lossSlots = 1 << 16 + lossWords = lossSlots / 64 +) + +type lossWindow struct { + mu sync.Mutex + base uint64 + inited bool + bits []uint64 + lost atomic.Uint64 + late atomic.Uint64 + _ [16]byte +} + +func newLossWindows(n int) []lossWindow { + w := make([]lossWindow, n) + for i := range w { + w[i].bits = make([]uint64, lossWords) + } + return w +} + +// A sequence number is only judged once it falls out of the window, so frames +// still queued in another rx worker are never miscounted as lost. +func (w *lossWindow) observe(seq uint64) { + w.mu.Lock() + if !w.inited { + // Start half a window below the first sequence seen, so frames another + // rx worker is still holding land inside the window rather than late. + if seq > lossSlots/2 { + w.base = seq - lossSlots/2 + } + w.inited = true + } + if seq < w.base { + w.mu.Unlock() + w.late.Add(1) + return + } + if seq >= w.base+lossSlots { + w.evict(seq - lossSlots + 1) + } + idx := seq & (lossSlots - 1) + w.bits[idx>>6] |= 1 << (idx & 63) + w.mu.Unlock() +} + +func (w *lossWindow) evict(newBase uint64) { + span := newBase - w.base + if span >= lossSlots { + var missing uint64 + for i := range w.bits { + missing += uint64(64 - bits.OnesCount64(w.bits[i])) + w.bits[i] = 0 + } + w.lost.Add(missing + span - lossSlots) + w.base = newBase + return + } + var missing uint64 + for s := w.base; s < newBase; s++ { + idx := s & (lossSlots - 1) + word, bit := idx>>6, uint64(1)<<(idx&63) + if w.bits[word]&bit == 0 { + missing++ + continue + } + w.bits[word] &^= bit + } + w.lost.Add(missing) + w.base = newBase +} diff --git a/main.go b/main.go index 0cb9622..3c2f46b 100644 --- a/main.go +++ b/main.go @@ -47,23 +47,21 @@ type direction struct { spec *frameSpec txStats []*txStats rxStats []*rxStats - streams []streamState + streams []lossWindow txFDs []int rxFDs []int reports chan string - prev sample - drops uint64 - nicTX nicCounters - nicRX nicCounters - nicTX0 nicCounters - nicRX0 nicCounters + prev sample + drops uint64 + nicTX nicCounters + nicRX nicCounters } type sample struct { txFrames, txBytes uint64 rxFrames, rxBytes uint64 - expected, count uint64 + lost, late uint64 crcErr, badMagic uint64 badLen uint64 txErrs, txShort uint64 @@ -165,13 +163,8 @@ func (d *direction) snapshot() sample { s.badLen += r.badLen.Load() } for i := range d.streams { - expected := d.streams[i].maxSeq.Load() + 1 - c := d.streams[i].count.Load() - if c == 0 { - continue - } - s.count += c - s.expected += expected + s.lost += d.streams[i].lost.Load() + s.late += d.streams[i].late.Load() } return s } @@ -187,19 +180,21 @@ func gbps(bytes, frames uint64, secs float64) float64 { } var intervalCols = []colSpec{ - {title: "TIME", width: 6, right: true}, + {title: "UPTIME", width: 9, right: true}, {title: "DIR", width: 5}, {title: "TX pps", width: 9, right: true}, {title: "TX Gb/s", width: 7, right: true}, {title: "RX pps", width: 9, right: true}, {title: "RX Gb/s", width: 7, right: true}, - {title: "GAP", width: 9, right: true}, + {title: "LOST", width: 8, right: true}, + {title: "LATE", width: 6, right: true}, {title: "CRC", width: 5, right: true}, {title: "BADMAG", width: 6, right: true}, {title: "KDROP", width: 7, right: true}, + {title: "ERRORS", width: 10, right: true}, } -func (d *direction) intervalRow(elapsed, secs, target float64) []string { +func (d *direction) intervalRow(elapsed time.Duration, secs, target float64) []string { now := d.snapshot() p := d.prev d.prev = now @@ -211,18 +206,21 @@ func (d *direction) intervalRow(elapsed, secs, target float64) []string { before := d.drops d.sampleDrops() + total := now.lost + now.crcErr + now.badMagic + now.badLen + d.drops return []string{ - fmt.Sprintf("%.0fs", elapsed), + uptime(elapsed), paint(d.short, cCyan), commas(uint64(float64(txF) / secs)), rateCell(gbps(txB, txF, secs), target), commas(uint64(float64(rxF) / secs)), rateCell(gbps(rxB, rxF, secs), target), - gapCell(int64(now.expected) - int64(now.count)), + statusCell(now.lost - p.lost), + statusCell(now.late - p.late), statusCell(now.crcErr - p.crcErr), statusCell(now.badMagic - p.badMagic), statusCell(d.drops - before), + statusCell(total), } } @@ -241,17 +239,6 @@ func (d *direction) reportNIC() []string { return parts } -func (d *direction) nicRunTotals() string { - var parts []string - if s := readNIC(d.tx.name).diff(d.nicTX0); s != "" { - parts = append(parts, d.tx.name+" tx-side: "+s) - } - if s := readNIC(d.rx.name).diff(d.nicRX0); s != "" { - parts = append(parts, d.rx.name+" rx-side: "+s) - } - return strings.Join(parts, "; ") -} - func buildDirection(label string, tx, rx endpoint, patIdx int, sizes []int, cfg config) (*direction, error) { d := &direction{ label: label, @@ -259,13 +246,11 @@ func buildDirection(label string, tx, rx endpoint, patIdx int, sizes []int, cfg tx: tx, rx: rx, spec: newFrameSpec(patIdx, rx.mac, tx.mac, sizes), - streams: make([]streamState, cfg.txWorkers), + streams: newLossWindows(cfg.txWorkers), reports: make(chan string, 64), } d.nicTX = readNIC(tx.name) d.nicRX = readNIC(rx.name) - d.nicTX0 = d.nicTX - d.nicRX0 = d.nicRX fm, err := fanoutMode(cfg.fanout, cfg.rxWorkers) if err != nil { @@ -350,13 +335,11 @@ func main() { var ( aName = flag.String("a", "", "first interface") bName = flag.String("b", "", "second interface") - duration = flag.Duration("duration", 0, "run time, 0 for until interrupted") sizesArg = flag.String("sizes", "64,128,256,512,1024,1280,1514", "frame sizes in bytes, excluding FCS, cycled per packet") patArg = flag.String("pattern", "prbs", "payload pattern") txN = flag.Int("tx", 4, "tx workers per direction") rxN = flag.Int("rx", 4, "rx workers per direction") batch = flag.Int("batch", 64, "frames per sendmmsg/recvmmsg call") - interval = flag.Duration("interval", time.Second, "report interval") fanout = flag.String("fanout", "lb", "rx fanout mode: none, hash, lb, cpu, rollover") duplex = flag.Bool("duplex", true, "run both directions simultaneously") txCPUs = flag.String("txcpus", "", "comma-separated CPUs to pin tx workers to") @@ -365,16 +348,16 @@ func main() { flag.Parse() if err := run(*aName, *bName, *sizesArg, *patArg, *fanout, *txCPUs, *rxCPUs, - *duration, *interval, *txN, *rxN, *batch, *duplex); err != nil { + *txN, *rxN, *batch, *duplex); err != nil { fmt.Fprintln(os.Stderr, "error:", err) os.Exit(1) } } -const drainTime = 500 * time.Millisecond +const reportInterval = 500 * time.Millisecond func run(aName, bName, sizesArg, patArg, fanout, txCPUsArg, rxCPUsArg string, - duration, interval time.Duration, txN, rxN, batch int, duplex bool) error { + txN, rxN, batch int, duplex bool) error { if aName == "" || bName == "" { return fmt.Errorf("both -a and -b are required") @@ -495,8 +478,6 @@ func run(aName, bName, sizesArg, patArg, fanout, txCPUsArg, rxCPUsArg string, humanBytes(uint64(sockBufSize(dirs[0].txFDs[0], unix.SO_SNDBUF))), humanBytes(uint64(sockBufSize(dirs[0].rxFDs[0], unix.SO_RCVBUF))))}, })) - fmt.Println(paint("GAP counts frames below the high-water mark not yet drained from the rx queues;", cDim)) - fmt.Println(paint("it is provisional and settles at the drain. RESULTS at the end is authoritative.", cDim)) fmt.Println() var doneTx, doneRx atomic.Bool @@ -516,29 +497,22 @@ func run(aName, bName, sizesArg, patArg, fanout, txCPUsArg, rxCPUsArg string, start := time.Now() close(startTx) - tick := time.NewTicker(interval) + tick := time.NewTicker(reportInterval) defer tick.Stop() - var deadline <-chan time.Time - if duration > 0 { - t := time.NewTimer(duration) - defer t.Stop() - deadline = t.C - } - last := time.Now() stats := &streamTable{cols: intervalCols, headerEvery: 20} -loop: for { select { case <-sig: - break loop - case <-deadline: - break loop + doneTx.Store(true) + doneRx.Store(true) + wg.Wait() + return nil case now := <-tick.C: secs := now.Sub(last).Seconds() last = now - elapsed := now.Sub(start).Seconds() + elapsed := now.Sub(start) for _, d := range dirs { for _, line := range stats.emit(d.intervalRow(elapsed, secs, target)) { fmt.Println(line) @@ -558,72 +532,4 @@ loop: } } } - - doneTx.Store(true) - elapsed := time.Since(start).Seconds() - time.Sleep(drainTime) - doneRx.Store(true) - wg.Wait() - - fmt.Println() - var resultRows, nicRows [][]string - var problems []string - for _, d := range dirs { - d.sampleDrops() - s := d.snapshot() - lost := int64(s.expected) - int64(s.count) - lostPct := 100 * float64(lost) / float64(max(s.expected, 1)) - - resultRows = append(resultRows, []string{ - paint(d.short, cCyan), - commas(s.txFrames), - commas(s.rxFrames), - humanBytes(s.rxBytes), - rateCell(gbps(s.txBytes, s.txFrames, elapsed), target), - rateCell(gbps(s.rxBytes, s.rxFrames, elapsed), target), - lostCell(lost), - fmt.Sprintf("%.4f", lostPct), - statusCell(s.crcErr), - statusCell(s.badMagic), - statusCell(s.badLen), - statusCell(d.drops), - }) - - if n := d.nicRunTotals(); n != "" { - nicRows = append(nicRows, []string{paint(d.short, cCyan), n}) - } - if lost != 0 { - problems = append(problems, fmt.Sprintf("%s lost %d frames (%.4f%%)", d.short, lost, lostPct)) - } - if s.crcErr != 0 { - problems = append(problems, fmt.Sprintf("%s had %d payload CRC failures", d.short, s.crcErr)) - } - if d.drops != 0 { - problems = append(problems, fmt.Sprintf("%s dropped %d frames in the kernel (host too slow, not the cable)", d.short, d.drops)) - } - } - - fmt.Println(renderBox(fmt.Sprintf("RESULTS after %.1fs", elapsed), - []string{"DIR", "TX FRAMES", "RX FRAMES", "RX DATA", "TX Gb/s", "RX Gb/s", "LOST", "LOST %", "CRC", "BAD", "BADLEN", "KDROP"}, - []bool{false, true, true, true, true, true, true, true, true, true, true, true}, - resultRows)) - - if len(nicRows) > 0 { - fmt.Println(renderBox("NIC COUNTER CHANGES DURING RUN", - []string{"DIR", "COUNTERS"}, []bool{false, false}, nicRows)) - problems = append(problems, "NIC error counters moved during the run") - } - - fmt.Println() - if len(problems) == 0 { - fmt.Println(paint(" PASS ", cBold+"\x1b[42m\x1b[30m") + " " + - paint(fmt.Sprintf("every frame arrived, no errors, %s each way at %.2f Gb/s", - humanBytes(dirs[0].snapshot().rxBytes), target), cGreen)) - } else { - fmt.Println(paint(" FAIL ", cBold+"\x1b[41m\x1b[37m")) - for _, p := range problems { - fmt.Println(paint(" ✗ "+p, cRed)) - } - } - return nil } diff --git a/render.go b/render.go index b6d2f33..e96eca1 100644 --- a/render.go +++ b/render.go @@ -3,6 +3,7 @@ package main import ( "fmt" "strings" + "time" ) const ( @@ -56,6 +57,11 @@ func pad(s string, w int, right bool) string { return s + strings.Repeat(" ", gap) } +func uptime(d time.Duration) string { + total := int(d.Seconds()) + return fmt.Sprintf("%d:%02d:%02d", total/3600, (total/60)%60, total%60) +} + func commas(v uint64) string { s := fmt.Sprintf("%d", v) if len(s) <= 3 { @@ -182,20 +188,6 @@ func statusCell(v uint64) string { return paint(s, cRed) } -func gapCell(v int64) string { - if v <= 0 { - return paint("0", cGreen) - } - return paint(commas(uint64(v)), cYellow) -} - -func lostCell(v int64) string { - if v <= 0 { - return paint("0", cGreen) - } - return paint(commas(uint64(v)), cRed) -} - func rateCell(gb float64, target float64) string { s := fmt.Sprintf("%.2f", gb) switch { diff --git a/rx.go b/rx.go index 04adf3a..33511e6 100644 --- a/rx.go +++ b/rx.go @@ -18,19 +18,13 @@ type rxStats struct { _ [24]byte } -type streamState struct { - maxSeq atomic.Uint64 - count atomic.Uint64 - _ [48]byte -} - type rxWorker struct { fd int batch int cpu int spec *frameSpec stats *rxStats - streams []streamState + streams []lossWindow reports chan string ready *sync.WaitGroup } @@ -68,14 +62,7 @@ func (w *rxWorker) run(done *atomic.Bool) { w.stats.bytes.Add(uint64(len(buf))) if int(p.stream) < len(w.streams) { - st := &w.streams[p.stream] - st.count.Add(1) - for { - old := st.maxSeq.Load() - if p.seq <= old || st.maxSeq.CompareAndSwap(old, p.seq) { - break - } - } + w.streams[p.stream].observe(p.seq) } if p.payLen > len(w.spec.ref) {