Stream forever with eviction-based per-stream loss detection

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
+}

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
+	prev  sample
-	drops  uint64
+	drops uint64
-	nicTX  nicCounters
+	nicTX nicCounters
-	nicRX  nicCounters
+	nicRX nicCounters
-	nicTX0 nicCounters
-	nicRX0 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
+		s.lost += d.streams[i].lost.Load()
-		c := d.streams[i].count.Load()
+		s.late += d.streams[i].late.Load()
-		if c == 0 {
-			continue
-		}
-		s.count += c
-		s.expected += expected
 	}
 	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
+			doneTx.Store(true)
-		case <-deadline:
+			doneRx.Store(true)
-			break loop
+			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
 }

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 {

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]
+				w.streams[p.stream].observe(p.seq)
-				st.count.Add(1)
-				for {
-					old := st.maxSeq.Load()
-					if p.seq <= old || st.maxSeq.CompareAndSwap(old, p.seq) {
-						break
-					}
-				}
 			}
 
 			if p.payLen > len(w.spec.ref) {