Measure cable length from hardware transmit and receive timestamps

main.go

@@ -46,6 +46,11 @@ type direction struct {
 	rxFDs   []int
 	reports chan string
 
+	probeSpec *frameSpec
+	probeTxFD int
+	probeRxFD int
+	cable     *cableStats
+
 	prevConsole sample
 	win         *rateWindow
 	est         *rateEstimators
@@ -259,6 +264,7 @@ func (d *direction) reset() {
 	d.dropBase = d.drops
 	d.heldFrames = heldValue{}
 	d.heldSent = heldValue{}
+	d.cable.reset()
 }
 
 // Returns the new start time, so the uptime shown alongside the totals counts
@@ -295,6 +301,8 @@ var intervalCols = []colSpec{
 	{title: "BADMAG", width: 7, right: true},
 	{title: "KDROP", width: 11, right: true},
 	{title: "ERRORS", width: 11, 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
@@ -307,6 +315,7 @@ type view struct {
 	lost, late       uint64
 	crc, badMagic    uint64
 	kdrop, errors    uint64
+	cable            cableView
 }
 
 // Cumulative fields, which need no rate window and are identical for both the
@@ -323,6 +332,7 @@ 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
 	return v
@@ -375,7 +385,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),
@@ -389,6 +399,8 @@ func (d *direction) row(elapsed time.Duration, v view, target float64) []string
 		statusCell(v.badMagic),
 		statusCell(v.kdrop),
 		statusCell(v.errors),
+		paint(v.cable.minText(), cCyan),
+		paint(length, cCyan),
 	}
 }
 
@@ -437,6 +449,28 @@ func buildDirection(label string, tx, rx endpoint, patIdx int, sizes []int, cfg
 		d.rxFDs = append(d.rxFDs, fd)
 		d.rxStats = append(d.rxStats, &rxStats{})
 	}
+
+	// The probe carries its own ethertype so it lands on a socket of its own, but
+	// it is left unsteered: it is a few frames a second and does not need a queue
+	// to itself, and the stamps are taken at the wire either way.
+	d.cable = newCableStats()
+	d.probeSpec = newFrameSpec(patIdx, 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
 	return d, nil
 }
 
@@ -472,6 +506,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 +529,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
+	streams    int
-	batch   int
+	batch      int
+	probeEther uint16
+	zeroNS     float64
+	nsPerM     float64
 }
 
 func main() {
@@ -497,11 +550,13 @@ func main() {
 		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", 4327.5, "both directions summed at zero cable length; belongs to the media adapters, recalibrate when they change")
+		nsPerM   = flag.Float64("ns-per-m", 10.909, "both directions summed, per metre of cable")
 	)
 	flag.Parse()
 
 	if err := run(*aName, *bName, *sizesArg, *patArg,
-		*streams, *batch, *duplex); err != nil {
+		*streams, *batch, *duplex, *zeroNS, *nsPerM); err != nil {
 		fmt.Fprintln(os.Stderr, "error:", err)
 		os.Exit(1)
 	}
@@ -519,7 +574,7 @@ const (
 )
 
 func run(aName, bName, sizesArg, patArg string,
-	nStreams, batch int, duplex bool) error {
+	nStreams, batch int, duplex bool, zeroNS, nsPerM float64) error {
 
 	if aName == "" || bName == "" {
 		return fmt.Errorf("both -a and -b are required")
@@ -575,8 +630,11 @@ func run(aName, bName, sizesArg, patArg string,
 	}
 
 	cfg := config{
-		streams: nStreams,
+		streams:    nStreams,
-		batch:   batch,
+		batch:      batch,
+		probeEther: uint16(etherBase + nStreams),
+		zeroNS:     zeroNS,
+		nsPerM:     nsPerM,
 	}
 
 	var dirs []*direction
@@ -626,6 +684,8 @@ func run(aName, bName, sizesArg, patArg string,
 				nStreams, ethertypes[0], ethertypes[len(ethertypes)-1])},
 			{"batch", fmt.Sprintf("%d frames per syscall", batch)},
 			{"payload verify", "crc32c on every frame"},
+			{"cable probe", fmt.Sprintf("%d-byte frame on ethertype 0x%04x every %s, hardware stamped at both macs",
+				probeSize, cfg.probeEther, probeInterval)},
 			{"duplex", fmt.Sprintf("%v", duplex)},
 			{"socket buffers", fmt.Sprintf("sndbuf %s, rcvbuf %s (granted)",
 				humanBytes(uint64(sockBufSize(dirs[0].txFDs[0], unix.SO_SNDBUF))),
@@ -639,7 +699,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 +732,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,14 +754,22 @@ 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 {
+			// Length needs both directions, so every row is sampled before any of
-				v := d.view(&d.prevConsole, secs)
+			// them is printed.
-				for _, line := range stats.emit(d.row(elapsed, v, target)) {
+			for i, d := range dirs {
+				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() {

probe.go

@@ -0,0 +1,243 @@
+package main
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+const (
+	// A stream number no data stream can take, so a probe is never mistaken for
+	// payload if one lands on the wrong socket.
+	probeStream = 0xffff
+	probeSize   = 64
+
+	// 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 the
+// difference is the two phys plus the cable and nothing else: 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)
+}
+
+// Summing both 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) != 2 {
+		return 0, false
+	}
+	var sum float64
+	for _, v := range views {
+		if v.cable.samples == 0 {
+			return 0, false
+		}
+		sum += float64(v.cable.min)
+	}
+	return (sum - 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),
+	}
+}
+
+// The two halves are produced by different goroutines in either order, so each
+// deposits its stamp and whichever lands second completes the pair.
+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()
+}
+
+// Sends one small frame at a time and collects its transmit stamp from the
+// socket's error queue before sending the next.
+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)
+	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 a stamp that
+		// arrived after its probe gave up would be handed to this one. Discard
+		// anything left over before sending.
+		for {
+			if _, _, _, _, err := unix.Recvmsg(p.fd, scratch, oob,
+				unix.MSG_ERRQUEUE|unix.MSG_DONTWAIT); err != nil {
+				break
+			}
+		}
+
+		putHeader(buf, p.spec.patIdx, probeStream, seq, probeSize-minFrame,
+			p.spec.crcFor[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])
+	}
+}
+
+// Reads probes on the far interface, where every frame carries a receive stamp
+// from the MAC.
+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)

system.go

@@ -196,7 +196,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 +284,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)
 	}
@@ -482,6 +483,7 @@ func configureSystem(ifnames []string, ethertypes []uint16) []checkResult {
 			}
 			out = append(out, checkCarrier(ifname, carrierWait))
 			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,110 @@
+package main
+
+import (
+	"fmt"
+	"unsafe"
+
+	"golang.org/x/sys/unix"
+)
+
+const (
+	hwtstampTxOn      = 1
+	hwtstampFilterAll = 1
+)
+
+// Read and written through the ifreq data pointer.
+type hwtstampConfig struct {
+	flags    int32
+	txType   int32
+	rxFilter int32
+}
+
+// Hardware timestamping is a property of the interface, not the socket: the MAC
+// has to be told to stamp on transmit and on receive before any socket can ask
+// for the values. 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
+}
+
+// The socket asks for the values the MAC is now producing. Only the raw
+// hardware clock is wanted; the software stamps would just be more control
+// message to copy.
+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)
+}
+
+// SCM_TIMESTAMPING carries three timespecs and the third is the raw hardware
+// clock; the first two are software clocks we did not ask for and which arrive
+// zeroed. A zero hardware stamp means the MAC did not produce one.
+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
+}

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,13 @@ func (d *display) render(dirs []*direction, views []view, elapsed time.Duration,
 	fb.flush()
 }
 
-func (d *display) section(y int, title string) int {
+// The right-hand annotation sits above the table's last column so it lines up
+// with the numbers under it rather than with the rule.
+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