diff --git a/main.go b/main.go index be1dafd..96a236b 100644 --- a/main.go +++ b/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 - batch int + streams 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, - batch: batch, + streams: nStreams, + 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 { - 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 { + 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() { diff --git a/probe.go b/probe.go new file mode 100644 index 0000000..e522867 --- /dev/null +++ b/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) + } +} diff --git a/render.go b/render.go index 8b1d221..1f41f39 100644 --- a/render.go +++ b/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) diff --git a/system.go b/system.go index 0460544..ab7c2c6 100644 --- a/system.go +++ 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 diff --git a/ts.go b/ts.go new file mode 100644 index 0000000..c7f9cce --- /dev/null +++ b/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 +} diff --git a/ui.go b/ui.go index 825473a..c4075ce 100644 --- a/ui.go +++ b/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