cabletest/probe.go

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