Cycle every payload pattern per packet and always run both directions

counters.go

@@ -1,27 +1,30 @@
 package main
 
 import (
-	"fmt"
 	"os"
 	"strconv"
 	"strings"
 )
 
-var nicFields = []string{
-	"rx_packets", "tx_packets",
-	"rx_errors", "tx_errors",
-	"rx_dropped", "tx_dropped",
-	"rx_crc_errors", "rx_missed_errors",
-	"rx_length_errors", "rx_over_errors",
-	"rx_frame_errors", "rx_fifo_errors",
-	"collisions",
-}
+// Split by direction of travel so each counter belongs to exactly one
+// direction: the transmitting interface owns the tx fields and the receiving one
+// the rx fields. Reading both sets off both interfaces double counts every drop.
+var (
+	nicTxFields = []string{
+		"tx_errors", "tx_dropped", "tx_fifo_errors",
+		"tx_carrier_errors", "tx_aborted_errors", "tx_window_errors",
+		"collisions",
+	}
+	nicRxFields = []string{
+		"rx_errors", "rx_dropped", "rx_crc_errors", "rx_missed_errors",
+		"rx_length_errors", "rx_over_errors", "rx_frame_errors", "rx_fifo_errors",
+	}
+)
 
 type nicCounters struct {
-	stats          map[string]uint64
-	carrierChanges uint64
-	carrierUp      uint64
-	carrierDown    uint64
+	tx          uint64
+	rx          uint64
+	carrierDown uint64
 }
 
 func readUint(path string) (uint64, bool) {
@@ -36,36 +39,22 @@ func readUint(path string) (uint64, bool) {
 	return v, true
 }
 
-func readNIC(ifname string) nicCounters {
-	c := nicCounters{stats: make(map[string]uint64, len(nicFields))}
-	base := "/sys/class/net/" + ifname
-	for _, f := range nicFields {
+func sumFields(base string, fields []string) uint64 {
+	var total uint64
+	for _, f := range fields {
 		if v, ok := readUint(base + "/statistics/" + f); ok {
-			c.stats[f] = v
+			total += v
 		}
 	}
-	c.carrierChanges, _ = readUint(base + "/carrier_changes")
-	c.carrierUp, _ = readUint(base + "/carrier_up_count")
+	return total
+}
+
+func readNIC(ifname string) nicCounters {
+	base := "/sys/class/net/" + ifname
+	c := nicCounters{
+		tx: sumFields(base, nicTxFields),
+		rx: sumFields(base, nicRxFields),
+	}
 	c.carrierDown, _ = readUint(base + "/carrier_down_count")
 	return c
 }
-
-func (c nicCounters) diff(prev nicCounters) string {
-	var parts []string
-	for _, f := range nicFields {
-		now, ok := c.stats[f]
-		if !ok {
-			continue
-		}
-		if d := now - prev.stats[f]; d != 0 && !strings.HasSuffix(f, "_packets") {
-			parts = append(parts, fmt.Sprintf("%s=%d", f, d))
-		}
-	}
-	if d := c.carrierChanges - prev.carrierChanges; d != 0 {
-		parts = append(parts, fmt.Sprintf("carrier_changes=%d", d))
-	}
-	if d := c.carrierDown - prev.carrierDown; d != 0 {
-		parts = append(parts, fmt.Sprintf("carrier_down=%d", d))
-	}
-	return strings.Join(parts, " ")
-}

frame.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"encoding/binary"
-	"fmt"
 	"hash/crc32"
 	"math/rand/v2"
 )
@@ -61,60 +60,60 @@ var patterns = []pattern{
 	}},
 }
 
-func patternIndex(name string) (int, error) {
-	for i, p := range patterns {
-		if p.name == name {
-			return i, nil
-		}
-	}
+func patternNames() []string {
 	names := make([]string, len(patterns))
 	for i, p := range patterns {
 		names[i] = p.name
 	}
-	return 0, fmt.Errorf("unknown pattern %q (have %v)", name, names)
+	return names
 }
 
+// Every pattern is laid out at full length up front with its checksum at each
+// size, so a sender picks a pattern by choosing a buffer, never by writing one.
 type frameSpec struct {
-	patIdx    int
-	ref       []byte
-	crcFor    map[int]uint32
+	refs      [][]byte
+	crcFor    []map[int]uint32
 	dstMAC    [6]byte
 	srcMAC    [6]byte
 	etherType uint16
 	sizes     []int
 	maxSize   int
+	maxPay    int
 }
 
-func newFrameSpec(patIdx int, dst, src [6]byte, etherType uint16, sizes []int) *frameSpec {
+func newFrameSpec(dst, src [6]byte, etherType uint16, sizes []int) *frameSpec {
 	maxSize := 0
 	for _, s := range sizes {
 		if s > maxSize {
 			maxSize = s
 		}
 	}
-	ref := make([]byte, maxSize-minFrame)
-	patterns[patIdx].fill(ref)
-	crcFor := make(map[int]uint32, len(sizes))
-	for _, s := range sizes {
-		crcFor[s] = crc32.Checksum(ref[:s-minFrame], crcTable)
-	}
-	return &frameSpec{
-		patIdx:    patIdx,
-		ref:       ref,
-		crcFor:    crcFor,
+	f := &frameSpec{
 		dstMAC:    dst,
 		srcMAC:    src,
 		etherType: etherType,
 		sizes:     sizes,
 		maxSize:   maxSize,
+		maxPay:    maxSize - minFrame,
 	}
+	for _, p := range patterns {
+		ref := make([]byte, f.maxPay)
+		p.fill(ref)
+		crcFor := make(map[int]uint32, len(sizes))
+		for _, s := range sizes {
+			crcFor[s] = crc32.Checksum(ref[:s-minFrame], crcTable)
+		}
+		f.refs = append(f.refs, ref)
+		f.crcFor = append(f.crcFor, crcFor)
+	}
+	return f
 }
 
-func (f *frameSpec) prefill(buf []byte) {
+func (f *frameSpec) prefill(buf []byte, patIdx int) {
 	copy(buf[0:6], f.dstMAC[:])
 	copy(buf[6:12], f.srcMAC[:])
 	binary.BigEndian.PutUint16(buf[12:14], f.etherType)
-	copy(buf[minFrame:], f.ref)
+	copy(buf[minFrame:], f.refs[patIdx])
 }
 
 func putHeader(buf []byte, patIdx int, stream uint16, seq uint64, payLen int, crc uint32) {
@@ -156,22 +155,3 @@ func parseHeader(buf []byte) (parsed, bool) {
 	}
 	return p, true
 }
-
-func firstDiff(got, want []byte) (int, int) {
-	n := len(got)
-	if len(want) < n {
-		n = len(want)
-	}
-	for i := 0; i < n; i++ {
-		if got[i] != want[i] {
-			bits := 0
-			x := got[i] ^ want[i]
-			for x != 0 {
-				bits += int(x & 1)
-				x >>= 1
-			}
-			return i, bits
-		}
-	}
-	return -1, 0
-}

main.go

@@ -44,7 +44,11 @@ type direction struct {
 	streams []lossWindow
 	txFDs   []int
 	rxFDs   []int
-	reports chan string
+
+	probeSpec *frameSpec
+	probeTxFD int
+	probeRxFD int
+	cable     *cableStats
 
 	prevConsole sample
 	win         *rateWindow
@@ -54,8 +58,8 @@ type direction struct {
 	drops       uint64
 	errBase     sample
 	dropBase    uint64
-	nicTX       nicCounters
-	nicRX       nicCounters
+	nicNow      uint64
+	nicBase     uint64
 }
 
 // Smoothing has to be steady against high-frequency noise yet still chase a
@@ -187,6 +191,7 @@ type sample struct {
 	crcErr, badMagic  uint64
 	badLen            uint64
 	txErrs, txShort   uint64
+	rxErrs            uint64
 }
 
 func lookupEndpoint(name string) (endpoint, error) {
@@ -242,6 +247,7 @@ func (d *direction) snapshot() sample {
 		s.crcErr += r.crcErr.Load()
 		s.badMagic += r.badMagic.Load()
 		s.badLen += r.badLen.Load()
+		s.rxErrs += r.rxErrs.Load()
 	}
 	for i := range d.streams {
 		s.lost += d.streams[i].lost.Load()
@@ -259,6 +265,8 @@ func (d *direction) reset() {
 	d.dropBase = d.drops
 	d.heldFrames = heldValue{}
 	d.heldSent = heldValue{}
+	d.nicBase = d.nicNow
+	d.cable.reset()
 }
 
 // Returns the new start time, so the uptime shown alongside the totals counts
@@ -294,7 +302,10 @@ var intervalCols = []colSpec{
 	{title: "CRC", width: 7, right: true},
 	{title: "BADMAG", width: 7, right: true},
 	{title: "KDROP", width: 11, right: true},
-	{title: "ERRORS", width: 11, right: true},
+	{title: "LINK", width: 13, right: true},
+	{title: "ERRORS", width: 13, right: true},
+	{title: "MIN ns", width: 9, right: true},
+	{title: "LEN m", width: 6, right: true},
 }
 
 // One interval's numbers, shared by the console table and the framebuffer so
@@ -306,7 +317,9 @@ type view struct {
 	rxFrames, rxGot  uint64
 	lost, late       uint64
 	crc, badMagic    uint64
-	kdrop, errors    uint64
+	kdrop, link      uint64
+	errors           uint64
+	cable            cableView
 }
 
 // Cumulative fields, which need no rate window and are identical for both the
@@ -323,8 +336,14 @@ func (d *direction) counters(now sample) view {
 		crc:      now.crcErr - b.crcErr,
 		badMagic: now.badMagic - b.badMagic,
 		kdrop:    d.drops - d.dropBase,
+		cable:    d.cable.view(),
 	}
-	v.errors = v.lost + v.crc + v.badMagic + (now.badLen - b.badLen) + v.kdrop
+	// A frame the stack refused and a frame the driver dropped are the same
+	// failure seen from either side of the ring, and never the same frame twice:
+	// a send that fails never reaches the driver to be dropped.
+	v.link = d.nicNow - d.nicBase +
+		(now.txErrs - b.txErrs) + (now.rxErrs - b.rxErrs)
+	v.errors = v.lost + v.crc + v.badMagic + (now.badLen - b.badLen) + v.kdrop + v.link
 	return v
 }
 
@@ -375,7 +394,7 @@ func (d *direction) displayView(t time.Time) view {
 	return v
 }
 
-func (d *direction) row(elapsed time.Duration, v view, target float64) []string {
+func (d *direction) row(elapsed time.Duration, v view, target float64, length string) []string {
 	return []string{
 		uptime(elapsed),
 		paint(d.short, cCyan),
@@ -388,40 +407,34 @@ func (d *direction) row(elapsed time.Duration, v view, target float64) []string
 		statusCell(v.crc),
 		statusCell(v.badMagic),
 		statusCell(v.kdrop),
+		statusCell(v.link),
 		statusCell(v.errors),
+		paint(v.cable.minText(), cCyan),
+		paint(length, cCyan),
 	}
 }
 
-func (d *direction) reportNIC() []string {
+// Sampled once a second, since these are sysfs reads; the display uses whatever
+// the last sample left behind.
+func (d *direction) sampleNIC() {
 	tx := readNIC(d.tx.name)
 	rx := readNIC(d.rx.name)
-	var parts []string
-	if s := tx.diff(d.nicTX); s != "" {
-		parts = append(parts, paint("  ▲ "+d.tx.name+" tx-side: "+s, cYellow))
-	}
-	if s := rx.diff(d.nicRX); s != "" {
-		parts = append(parts, paint("  ▲ "+d.rx.name+" rx-side: "+s, cYellow))
-	}
-	d.nicTX = tx
-	d.nicRX = rx
-	return parts
+	d.nicNow = tx.tx + rx.rx + tx.carrierDown
 }
 
-func buildDirection(label string, tx, rx endpoint, patIdx int, sizes []int, cfg config) (*direction, error) {
+func buildDirection(label string, tx, rx endpoint, sizes []int, cfg config) (*direction, error) {
 	d := &direction{
 		label:   label,
 		short:   tx.tag + "→" + rx.tag,
 		tx:      tx,
 		rx:      rx,
 		streams: newLossWindows(cfg.streams),
-		reports: make(chan string, 64),
+		cable:   newCableStats(),
 	}
-	d.nicTX = readNIC(tx.name)
-	d.nicRX = readNIC(rx.name)
 
 	for i := 0; i < cfg.streams; i++ {
 		et := uint16(etherBase + i)
-		d.specs = append(d.specs, newFrameSpec(patIdx, rx.mac, tx.mac, et, sizes))
+		d.specs = append(d.specs, newFrameSpec(rx.mac, tx.mac, et, sizes))
 
 		fd, err := openTxSocket(tx.idx)
 		if err != nil {
@@ -437,6 +450,30 @@ func buildDirection(label string, tx, rx endpoint, patIdx int, sizes []int, cfg
 		d.rxFDs = append(d.rxFDs, fd)
 		d.rxStats = append(d.rxStats, &rxStats{})
 	}
+
+	// Deliberately given no flow rule: a few frames a second does not need a
+	// queue of its own, and the stamps are taken at the wire either way.
+	d.probeSpec = newFrameSpec(rx.mac, tx.mac, cfg.probeEther, []int{probeSize})
+	fd, err := openTxSocket(tx.idx)
+	if err != nil {
+		return nil, fmt.Errorf("%s probe tx socket: %w", label, err)
+	}
+	if err := enableTxTimestamps(fd); err != nil {
+		return nil, fmt.Errorf("%s probe tx timestamps: %w", label, err)
+	}
+	d.probeTxFD = fd
+	fd, err = openRxSocket(rx.idx, cfg.probeEther)
+	if err != nil {
+		return nil, fmt.Errorf("%s probe rx socket: %w", label, err)
+	}
+	if err := enableRxTimestamps(fd); err != nil {
+		return nil, fmt.Errorf("%s probe rx timestamps: %w", label, err)
+	}
+	d.probeRxFD = fd
+
+	// Whatever the interfaces have counted before now is not ours.
+	d.sampleNIC()
+	d.nicBase = d.nicNow
 	return d, nil
 }
 
@@ -463,7 +500,6 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, cfg c
 			spec:    d.specs[i],
 			stats:   d.rxStats[i],
 			streams: d.streams,
-			reports: d.reports,
 			ready:   rxReady,
 		}
 		wg.Add(1)
@@ -472,6 +508,20 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, cfg c
 			w.run(doneRx)
 		}()
 	}
+
+	sender := &probeSender{fd: d.probeTxFD, spec: d.probeSpec, stats: d.cable}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		sender.run(doneTx, startTx)
+	}()
+
+	receiver := &probeReceiver{fd: d.probeRxFD, stats: d.cable, ready: rxReady}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		receiver.run(doneRx)
+	}()
 }
 
 func (d *direction) close() {
@@ -481,11 +531,16 @@ func (d *direction) close() {
 	for _, fd := range d.rxFDs {
 		unix.Close(fd)
 	}
+	unix.Close(d.probeTxFD)
+	unix.Close(d.probeRxFD)
 }
 
 type config struct {
-	streams int
-	batch   int
+	streams    int
+	batch      int
+	probeEther uint16
+	zeroNS     float64
+	nsPerM     float64
 }
 
 func main() {
@@ -493,15 +548,15 @@ func main() {
 		aName    = flag.String("a", "", "first interface")
 		bName    = flag.String("b", "", "second interface")
 		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")
 		streams  = flag.Int("streams", 7, "independent streams per direction, capped by rx rings; each gets its own ethertype, steered by a flow rule to its own rx queue")
 		batch    = flag.Int("batch", 64, "frames per sendmmsg/recvmmsg call")
-		duplex   = flag.Bool("duplex", true, "run both directions simultaneously")
+		zeroNS   = flag.Float64("zero-ns", 2163.77, "mean of both directions at zero cable length; belongs to the media adapters, recalibrate when they change")
+		nsPerM   = flag.Float64("ns-per-m", 5.4545, "mean of both directions, per metre of cable")
 	)
 	flag.Parse()
 
-	if err := run(*aName, *bName, *sizesArg, *patArg,
-		*streams, *batch, *duplex); err != nil {
+	if err := run(*aName, *bName, *sizesArg,
+		*streams, *batch, *zeroNS, *nsPerM); err != nil {
 		fmt.Fprintln(os.Stderr, "error:", err)
 		os.Exit(1)
 	}
@@ -518,8 +573,8 @@ const (
 	totalsHold     = 50 * time.Millisecond
 )
 
-func run(aName, bName, sizesArg, patArg string,
-	nStreams, batch int, duplex bool) error {
+func run(aName, bName, sizesArg string,
+	nStreams, batch int, zeroNS, nsPerM float64) error {
 
 	if aName == "" || bName == "" {
 		return fmt.Errorf("both -a and -b are required")
@@ -528,10 +583,6 @@ func run(aName, bName, sizesArg, patArg string,
 	if err != nil {
 		return err
 	}
-	patIdx, err := patternIndex(patArg)
-	if err != nil {
-		return err
-	}
 	a, err := lookupEndpoint(aName)
 	if err != nil {
 		return err
@@ -575,22 +626,20 @@ func run(aName, bName, sizesArg, patArg string,
 	}
 
 	cfg := config{
-		streams: nStreams,
-		batch:   batch,
+		streams:    nStreams,
+		batch:      batch,
+		probeEther: uint16(etherBase + nStreams),
+		zeroNS:     zeroNS,
+		nsPerM:     nsPerM,
 	}
 
 	var dirs []*direction
-	d0, err := buildDirection(a.name+"->"+b.name, a, b, patIdx, sizes, cfg)
-	if err != nil {
-		return err
-	}
-	dirs = append(dirs, d0)
-	if duplex {
-		d1, err := buildDirection(b.name+"->"+a.name, b, a, patIdx, sizes, cfg)
+	for _, p := range [][2]endpoint{{a, b}, {b, a}} {
+		d, err := buildDirection(p[0].name+"->"+p[1].name, p[0], p[1], sizes, cfg)
 		if err != nil {
 			return err
 		}
-		dirs = append(dirs, d1)
+		dirs = append(dirs, d)
 	}
 	defer func() {
 		for _, d := range dirs {
@@ -617,21 +666,19 @@ func run(aName, bName, sizesArg, patArg string,
 	for i, s := range sizes {
 		sizeStrs[i] = fmt.Sprintf("%d", s)
 	}
-	fmt.Println(renderBox("TEST",
+	fmt.Println(renderBox("CONFIG",
 		[]string{"SETTING", "VALUE"},
 		[]bool{false, false}, [][]string{
-			{"pattern", patterns[patIdx].name},
 			{"frame sizes", strings.Join(sizeStrs, " ")},
-			{"streams", fmt.Sprintf("%d per direction, ethertypes 0x%04x-0x%04x, one rx queue each",
+			{"streams", fmt.Sprintf("%d per direction, ethertypes 0x%04x-0x%04x",
 				nStreams, ethertypes[0], ethertypes[len(ethertypes)-1])},
+			{"probe", fmt.Sprintf("ethertype 0x%04x every %s", cfg.probeEther, probeInterval)},
 			{"batch", fmt.Sprintf("%d frames per syscall", batch)},
-			{"payload verify", "crc32c on every frame"},
-			{"duplex", fmt.Sprintf("%v", duplex)},
-			{"socket buffers", fmt.Sprintf("sndbuf %s, rcvbuf %s (granted)",
+			{"calibration", fmt.Sprintf("%g ns at zero length, %g ns/m", zeroNS, nsPerM)},
+			{"buffers", fmt.Sprintf("sndbuf %s, rcvbuf %s",
 				humanBytes(uint64(sockBufSize(dirs[0].txFDs[0], unix.SO_SNDBUF))),
 				humanBytes(uint64(sockBufSize(dirs[0].rxFDs[0], unix.SO_RCVBUF))))},
 		}))
-	fmt.Println(paint("rates are per interval; error counts are cumulative, press space to reset them", cDim))
 	fmt.Println()
 
 	var doneTx, doneRx atomic.Bool
@@ -639,7 +686,7 @@ func run(aName, bName, sizesArg, patArg string,
 	var rxReady sync.WaitGroup
 	startTx := make(chan struct{})
 	for _, d := range dirs {
-		rxReady.Add(len(d.rxFDs))
+		rxReady.Add(len(d.rxFDs) + 1)
 	}
 	for _, d := range dirs {
 		d.start(&wg, &doneTx, &doneRx, cfg, &rxReady, startTx)
@@ -672,6 +719,7 @@ func run(aName, bName, sizesArg, patArg string,
 
 	last := time.Now()
 	views := make([]view, len(dirs))
+	rows := make([]view, len(dirs))
 	for _, d := range dirs {
 		d.win = newRateWindow(int(rateWindowSpan/displayInterval) + 1)
 		d.est = newRateEstimators()
@@ -693,28 +741,25 @@ func run(aName, bName, sizesArg, patArg string,
 			for i, d := range dirs {
 				views[i] = d.displayView(now)
 			}
-			disp.render(dirs, views, now.Sub(start), target)
+			disp.render(dirs, views, now.Sub(start), target, cfg.cableText(views))
 		case now := <-tick.C:
 			secs := now.Sub(last).Seconds()
 			last = now
 			elapsed := now.Sub(start)
-			for _, d := range dirs {
-				v := d.view(&d.prevConsole, secs)
-				for _, line := range stats.emit(d.row(elapsed, v, target)) {
+			// Length needs both directions, so every row is sampled before any of
+			// them is printed.
+			for i, d := range dirs {
+				d.sampleNIC()
+				rows[i] = d.view(&d.prevConsole, secs)
+			}
+			length := "-"
+			if m, ok := cfg.cableMetres(rows); ok {
+				length = fmt.Sprintf("%.1f", m)
+			}
+			for i, d := range dirs {
+				for _, line := range stats.emit(d.row(elapsed, rows[i], target, length)) {
 					fmt.Println(line)
 				}
-				for _, line := range d.reportNIC() {
-					fmt.Println(line)
-				}
-				for {
-					select {
-					case msg := <-d.reports:
-						fmt.Println(paint("  ✗ "+d.short+": "+msg, cRed))
-						continue
-					default:
-					}
-					break
-				}
 			}
 		}
 	}

probe.go

@@ -0,0 +1,236 @@
+package main
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+const (
+	// Above any real stream number, so a probe is never taken for payload.
+	probeStream  = 0xffff
+	probeSize    = 64
+	probePattern = 0
+
+	// The mac has only a handful of transmit stamp slots. Asking faster than it
+	// can drain them gets slots recycled while a stamp is still outstanding, and
+	// the one that comes back then belongs to a different frame.
+	probeInterval = 200 * time.Millisecond
+	probeTimeout  = 20 * time.Millisecond
+	// A phy costs microseconds and a hundred metres of copper costs five hundred
+	// nanoseconds, so anything past this is a broken stamp, not a slow frame.
+	probeMaxDelay = 50 * time.Microsecond
+	probePendCap  = 256
+)
+
+// Both ports hang off one PTP clock, so a transmit stamp from one and a receive
+// stamp from the other subtract directly. Both are taken at the mac, so all host
+// time and all queueing falls outside the stamped interval, which is why load
+// does not move it.
+type cableStats struct {
+	mu      sync.Mutex
+	min     int64
+	samples uint64
+
+	txPend map[uint64]int64
+	rxPend map[uint64]int64
+}
+
+type cableView struct {
+	min     int64
+	samples uint64
+}
+
+func (v cableView) minText() string {
+	if v.samples == 0 {
+		return "-"
+	}
+	return commasInt(v.min)
+}
+
+// Averaging the two directions cancels the phy asymmetry between them, which is
+// about 790ns and swamps any cable, so one direction alone cannot give a length.
+func (c config) cableMetres(views []view) (float64, bool) {
+	if len(views) == 0 {
+		return 0, false
+	}
+	var sum float64
+	for _, v := range views {
+		if v.cable.samples == 0 {
+			return 0, false
+		}
+		sum += float64(v.cable.min)
+	}
+	mean := sum / float64(len(views))
+	return (mean - c.zeroNS) / c.nsPerM, true
+}
+
+func (c config) cableText(views []view) string {
+	m, ok := c.cableMetres(views)
+	if !ok {
+		return ""
+	}
+	return fmt.Sprintf("cable %.1f m", m)
+}
+
+func newCableStats() *cableStats {
+	return &cableStats{
+		txPend: make(map[uint64]int64, probePendCap),
+		rxPend: make(map[uint64]int64, probePendCap),
+	}
+}
+
+func (c *cableStats) put(seq uint64, ts int64, tx bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	mine, theirs := c.txPend, c.rxPend
+	if !tx {
+		mine, theirs = c.rxPend, c.txPend
+	}
+	other, ok := theirs[seq]
+	if !ok {
+		if len(mine) >= probePendCap {
+			clear(mine)
+		}
+		mine[seq] = ts
+		return
+	}
+	delete(theirs, seq)
+
+	delta := ts - other
+	if tx {
+		delta = -delta
+	}
+	// The driver rebuilds a full timestamp from a truncated hardware value plus a
+	// cached clock read, and a stale cache lands hundreds of milliseconds out. A
+	// minimum would latch onto the first of those and never recover.
+	if delta <= 0 || delta > int64(probeMaxDelay) {
+		return
+	}
+	if c.samples == 0 || delta < c.min {
+		c.min = delta
+	}
+	c.samples++
+}
+
+func (c *cableStats) view() cableView {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return cableView{c.min, c.samples}
+}
+
+func (c *cableStats) reset() {
+	c.mu.Lock()
+	c.min, c.samples = 0, 0
+	clear(c.txPend)
+	clear(c.rxPend)
+	c.mu.Unlock()
+}
+
+type probeSender struct {
+	fd    int
+	spec  *frameSpec
+	stats *cableStats
+}
+
+func (p *probeSender) run(done *atomic.Bool, startTx <-chan struct{}) {
+	buf := make([]byte, probeSize)
+	p.spec.prefill(buf, probePattern)
+	oob := make([]byte, 512)
+	scratch := make([]byte, 1)
+
+	<-startTx
+
+	tick := time.NewTicker(probeInterval)
+	defer tick.Stop()
+
+	var seq uint64
+	for !done.Load() {
+		<-tick.C
+
+		// Stamps are matched to sends by position in the queue, so one that
+		// arrived after its probe gave up would be handed to this probe.
+		for {
+			if _, _, _, _, err := unix.Recvmsg(p.fd, scratch, oob,
+				unix.MSG_ERRQUEUE|unix.MSG_DONTWAIT); err != nil {
+				break
+			}
+		}
+
+		putHeader(buf, probePattern, probeStream, seq, probeSize-minFrame,
+			p.spec.crcFor[probePattern][probeSize])
+		err := unix.Send(p.fd, buf, 0)
+		// The sequence advances even when a probe fails, so a stale receive half
+		// can never be paired with a later probe that reused its number.
+		cur := seq
+		seq++
+		if err != nil {
+			continue
+		}
+		ts, ok := p.awaitTx(scratch, oob)
+		if !ok {
+			continue
+		}
+		p.stats.put(cur, ts, true)
+	}
+}
+
+func (p *probeSender) awaitTx(scratch, oob []byte) (int64, bool) {
+	fds := []unix.PollFd{{Fd: int32(p.fd), Events: unix.POLLERR}}
+	deadline := time.Now().Add(probeTimeout)
+	for {
+		ms := int(time.Until(deadline).Milliseconds())
+		if ms <= 0 {
+			return 0, false
+		}
+		n, err := unix.Poll(fds, ms)
+		if err == unix.EINTR {
+			continue
+		}
+		if err != nil || n == 0 {
+			return 0, false
+		}
+		_, oobn, _, _, err := unix.Recvmsg(p.fd, scratch, oob,
+			unix.MSG_ERRQUEUE|unix.MSG_DONTWAIT)
+		if err == unix.EAGAIN || err == unix.EINTR {
+			continue
+		}
+		if err != nil {
+			return 0, false
+		}
+		return hwTimestamp(oob[:oobn])
+	}
+}
+
+type probeReceiver struct {
+	fd    int
+	stats *cableStats
+	ready *sync.WaitGroup
+}
+
+func (r *probeReceiver) run(done *atomic.Bool) {
+	buf := make([]byte, maxFrame)
+	oob := make([]byte, 512)
+
+	r.ready.Done()
+
+	for !done.Load() {
+		n, oobn, _, _, err := unix.Recvmsg(r.fd, buf, oob, 0)
+		if err != nil {
+			continue
+		}
+		h, ok := parseHeader(buf[:n])
+		if !ok || h.stream != probeStream {
+			continue
+		}
+		ts, ok := hwTimestamp(oob[:oobn])
+		if !ok {
+			continue
+		}
+		r.stats.put(h.seq, ts, false)
+	}
+}

render.go

@@ -75,6 +75,13 @@ func commas(v uint64) string {
 	return strings.Join(append([]string{s}, parts...), ",")
 }
 
+func commasInt(v int64) string {
+	if v < 0 {
+		return "-" + commas(uint64(-v))
+	}
+	return commas(uint64(v))
+}
+
 func humanBytes(b uint64) string {
 	const unit = 1000.0
 	v := float64(b)

rx.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"fmt"
 	"hash/crc32"
 	"sync"
 	"sync/atomic"
@@ -15,7 +14,8 @@ type rxStats struct {
 	badMagic atomic.Uint64
 	badLen   atomic.Uint64
 	crcErr   atomic.Uint64
-	_        [24]byte
+	rxErrs   atomic.Uint64
+	_        [16]byte
 }
 
 type rxWorker struct {
@@ -24,7 +24,6 @@ type rxWorker struct {
 	spec    *frameSpec
 	stats   *rxStats
 	streams []lossWindow
-	reports chan string
 	ready   *sync.WaitGroup
 }
 
@@ -44,7 +43,7 @@ func (w *rxWorker) run(done *atomic.Bool) {
 		n, err := recvmmsg(w.fd, hdrs, unix.MSG_WAITFORONE)
 		if n <= 0 {
 			if err != nil && err != unix.EAGAIN && err != unix.EINTR {
-				w.report(fmt.Sprintf("recvmmsg: %v", err))
+				w.stats.rxErrs.Add(1)
 			}
 			continue
 		}
@@ -62,25 +61,14 @@ func (w *rxWorker) run(done *atomic.Bool) {
 				w.streams[p.stream].observe(p.seq)
 			}
 
-			if p.payLen > len(w.spec.ref) {
+			if p.payLen > w.spec.maxPay {
 				w.stats.badLen.Add(1)
 				continue
 			}
 			pay := buf[minFrame : minFrame+p.payLen]
-			if crc32.Checksum(pay, crcTable) == p.crc {
-				continue
+			if crc32.Checksum(pay, crcTable) != p.crc {
+				w.stats.crcErr.Add(1)
 			}
-			w.stats.crcErr.Add(1)
-			off, bits := firstDiff(pay, w.spec.ref[:p.payLen])
-			w.report(fmt.Sprintf("payload mismatch stream=%d seq=%d len=%d first-diff-offset=%d bits=%d",
-				p.stream, p.seq, p.payLen, off, bits))
 		}
 	}
 }
-
-func (w *rxWorker) report(msg string) {
-	select {
-	case w.reports <- msg:
-	default:
-	}
-}

system.go

@@ -6,7 +6,6 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
-	"time"
 	"unsafe"
 
 	"golang.org/x/sys/unix"
@@ -196,7 +195,8 @@ func checkFlowRules(fd int, ifname string, ethertypes []uint16) checkResult {
 	return res
 }
 
-type ethtoolIfreq struct {
+// The ifreq shape used by every ioctl that passes its payload by pointer.
+type dataIfreq struct {
 	name [unix.IFNAMSIZ]byte
 	data unsafe.Pointer
 	_    [16]byte
@@ -283,7 +283,7 @@ func (r checkResult) detail() string {
 }
 
 func ethtoolCall(fd int, ifname string, data unsafe.Pointer) error {
-	var ifr ethtoolIfreq
+	var ifr dataIfreq
 	if len(ifname) >= unix.IFNAMSIZ {
 		return fmt.Errorf("interface name %q too long", ifname)
 	}
@@ -379,24 +379,6 @@ func checkLinkUp(fd int, ifname string) checkResult {
 	return res
 }
 
-func checkCarrier(ifname string, wait time.Duration) checkResult {
-	res := checkResult{item: ifname + " carrier"}
-	deadline := time.Now().Add(wait)
-	for {
-		v, ok := readUint("/sys/class/net/" + ifname + "/carrier")
-		if ok && v == 1 {
-			res.state = "present"
-			return res
-		}
-		if time.Now().After(deadline) {
-			res.err = fmt.Errorf("no carrier after %s", wait)
-			res.fatal = true
-			return res
-		}
-		time.Sleep(100 * time.Millisecond)
-	}
-}
-
 func checkCoalesce(fd int, ifname string, rxUsecs, txUsecs uint32) checkResult {
 	res := checkResult{item: ifname + " coalesce"}
 	ec, err := getCoalesce(fd, ifname)
@@ -429,18 +411,18 @@ func checkCoalesce(fd int, ifname string, rxUsecs, txUsecs uint32) checkResult {
 	return res
 }
 
-func checkRings(fd int, ifname string, rxWant, txWant uint32) (checkResult, bool) {
+func checkRings(fd int, ifname string, rxWant, txWant uint32) checkResult {
 	res := checkResult{item: ifname + " rings"}
 	rp, err := getRings(fd, ifname)
 	if err != nil {
 		res.err = err
-		return res, false
+		return res
 	}
 	rx := min(rxWant, rp.rxMaxPending)
 	tx := min(txWant, rp.txMaxPending)
 	if rp.rxPending == rx && rp.txPending == tx {
 		res.state = fmt.Sprintf("rx=%d tx=%d", rp.rxPending, rp.txPending)
-		return res, false
+		return res
 	}
 	was := fmt.Sprintf("rx=%d tx=%d", rp.rxPending, rp.txPending)
 	rp.cmd = unix.ETHTOOL_SRINGPARAM
@@ -449,11 +431,11 @@ func checkRings(fd int, ifname string, rxWant, txWant uint32) (checkResult, bool
 	if err := ethtoolCall(fd, ifname, unsafe.Pointer(&rp)); err != nil {
 		res.err = err
 		res.state = "could not set"
-		return res, false
+		return res
 	}
 	res.fixed = true
 	res.state = fmt.Sprintf("was %s, now rx=%d tx=%d (link reset)", was, rx, tx)
-	return res, true
+	return res
 }
 
 func withIoctlSocket(fn func(fd int) []checkResult) []checkResult {
@@ -474,14 +456,9 @@ func configureSystem(ifnames []string, ethertypes []uint16) []checkResult {
 
 			// Ring changes reprogram the queues, so flow rules pointing at those
 			// queues have to be installed afterwards.
-			carrierWait := 3 * time.Second
-			ringRes, ringReset := checkRings(fd, ifname, wantRxRing, wantTxRing)
-			out = append(out, ringRes)
-			if ringReset {
-				carrierWait = 10 * time.Second
-			}
-			out = append(out, checkCarrier(ifname, carrierWait))
+			out = append(out, checkRings(fd, ifname, wantRxRing, wantTxRing))
 			out = append(out, checkCoalesce(fd, ifname, wantCoalesceUsecs, wantCoalesceUsecs))
+			out = append(out, checkTimestamping(fd, ifname))
 			out = append(out, checkFlowRules(fd, ifname, ethertypes))
 		}
 		return out

ts.go

@@ -0,0 +1,103 @@
+package main
+
+import (
+	"fmt"
+	"unsafe"
+
+	"golang.org/x/sys/unix"
+)
+
+const (
+	hwtstampTxOn      = 1
+	hwtstampFilterAll = 1
+)
+
+type hwtstampConfig struct {
+	flags    int32
+	txType   int32
+	rxFilter int32
+}
+
+// Receive stamping is filtered by protocol and ours is not PTP, so nothing
+// narrower than "all" will see our frames.
+func checkTimestamping(fd int, ifname string) checkResult {
+	res := checkResult{item: ifname + " hw timestamps"}
+	desc := func(c hwtstampConfig) string {
+		return fmt.Sprintf("tx_type=%d rx_filter=%d", c.txType, c.rxFilter)
+	}
+
+	hwtstampCall := func(req uintptr, cfg *hwtstampConfig) error {
+		var ifr dataIfreq
+		copy(ifr.name[:], ifname)
+		ifr.data = unsafe.Pointer(cfg)
+		if _, _, errno := unix.Syscall(unix.SYS_IOCTL, uintptr(fd), req,
+			uintptr(unsafe.Pointer(&ifr))); errno != 0 {
+			return errno
+		}
+		return nil
+	}
+
+	var have hwtstampConfig
+	if err := hwtstampCall(unix.SIOCGHWTSTAMP, &have); err != nil {
+		res.err = err
+		res.fatal = true
+		return res
+	}
+	if have.txType == hwtstampTxOn && have.rxFilter == hwtstampFilterAll {
+		res.state = desc(have)
+		return res
+	}
+
+	// The ioctl reports back what the driver actually applied, which can be
+	// narrower than what was asked for.
+	want := hwtstampConfig{txType: hwtstampTxOn, rxFilter: hwtstampFilterAll}
+	if err := hwtstampCall(unix.SIOCSHWTSTAMP, &want); err != nil {
+		res.err = err
+		res.state = "could not set"
+		res.fatal = true
+		return res
+	}
+	if want.txType != hwtstampTxOn || want.rxFilter != hwtstampFilterAll {
+		res.err = fmt.Errorf("driver applied %s instead", desc(want))
+		res.fatal = true
+		return res
+	}
+	res.fixed = true
+	res.state = fmt.Sprintf("was %s, now %s", desc(have), desc(want))
+	return res
+}
+
+func enableTxTimestamps(fd int) error {
+	return unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_TIMESTAMPING,
+		unix.SOF_TIMESTAMPING_TX_HARDWARE|
+			unix.SOF_TIMESTAMPING_RAW_HARDWARE|
+			unix.SOF_TIMESTAMPING_OPT_TSONLY)
+}
+
+func enableRxTimestamps(fd int) error {
+	return unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_TIMESTAMPING,
+		unix.SOF_TIMESTAMPING_RX_HARDWARE|unix.SOF_TIMESTAMPING_RAW_HARDWARE)
+}
+
+// Three timespecs, of which the third is the raw hardware clock. A zero there
+// means the mac did not produce a stamp.
+const scmTimestampingLen = 3 * int(unsafe.Sizeof(unix.Timespec{}))
+
+func hwTimestamp(oob []byte) (int64, bool) {
+	msgs, err := unix.ParseSocketControlMessage(oob)
+	if err != nil {
+		return 0, false
+	}
+	for _, m := range msgs {
+		if m.Header.Level != unix.SOL_SOCKET || m.Header.Type != unix.SCM_TIMESTAMPING {
+			continue
+		}
+		if len(m.Data) < scmTimestampingLen {
+			return 0, false
+		}
+		ts := (*[3]unix.Timespec)(unsafe.Pointer(&m.Data[0]))
+		ns := ts[2].Nano()
+		return ns, ns != 0
+	}
+	return 0, false
+}

tx.go

@@ -24,10 +24,14 @@ type txWorker struct {
 }
 
 func (w *txWorker) run(done *atomic.Bool) {
+	// The batch is not a multiple of the pattern count, so the pairing of pattern
+	// to size rotates every pass rather than settling into a fixed one.
 	bufs := make([][]byte, w.batch)
+	pats := make([]int, w.batch)
 	for i := range bufs {
 		bufs[i] = make([]byte, w.spec.maxSize)
-		w.spec.prefill(bufs[i])
+		pats[i] = i % len(patterns)
+		w.spec.prefill(bufs[i], pats[i])
 	}
 	hdrs, iovs := newMmsghdrs(bufs)
 	sizes := make([]int, w.batch)
@@ -44,7 +48,8 @@ func (w *txWorker) run(done *atomic.Bool) {
 				si = 0
 			}
 			sizes[i] = size
-			putHeader(bufs[i], w.spec.patIdx, w.stream, seq+uint64(i), size-minFrame, w.spec.crcFor[size])
+			putHeader(bufs[i], pats[i], w.stream, seq+uint64(i), size-minFrame,
+				w.spec.crcFor[pats[i]][size])
 			iovs[i].Len = uint64(size)
 		}
 

ui.go

@@ -216,7 +216,7 @@ func rateColor(gb, target float64) rgb {
 	}
 }
 
-func (d *display) render(dirs []*direction, views []view, elapsed time.Duration, target float64) {
+func (d *display) render(dirs []*direction, views []view, elapsed time.Duration, target float64, cable string) {
 	fb := d.fb
 	fb.fill(uiBg)
 
@@ -248,7 +248,7 @@ func (d *display) render(dirs []*direction, views []view, elapsed time.Duration,
 	avail := fb.h - (bandY + bandH) - btnH - 2*uiMargin
 	y := bandY + bandH + 8 + (avail-2*sectionH-gap)/2
 
-	y = d.section(y, "RATE")
+	y = d.section(y, "RATE", "")
 	d.rightAt(colTxGbEnd, y, "TX Gb/s", uiDim)
 	d.rightAt(colTxPPSEnd, y, "TX pps", uiDim)
 	d.rightAt(colRxGbEnd, y, "RX Gb/s", uiDim)
@@ -265,7 +265,7 @@ func (d *display) render(dirs []*direction, views []view, elapsed time.Duration,
 	}
 
 	y += gap
-	y = d.section(y, "OVERALL")
+	y = d.section(y, "OVERALL", cable)
 	d.rightAt(colFramesEnd, y, "frames", uiDim)
 	d.rightAt(colDataEnd, y, "data", uiDim)
 	d.rightAt(colLostEnd, y, "lost", uiDim)
@@ -289,8 +289,11 @@ func (d *display) render(dirs []*direction, views []view, elapsed time.Duration,
 	fb.flush()
 }
 
-func (d *display) section(y int, title string) int {
+func (d *display) section(y int, title, right string) int {
 	d.small.draw(d.fb, uiMargin, y, title, uiFg)
+	if right != "" {
+		d.right(d.small, d.cellX(colKdropEnd), y, right, uiCyan)
+	}
 	ruleY := y + d.small.cellH + 3
 	d.fb.rect(uiMargin, ruleY, d.fb.w-2*uiMargin, 1, uiRule)
 	return ruleY + 7