From e77574712dc35873614bb6f405f400655dd069f1 Mon Sep 17 00:00:00 2001 From: flamingcow Date: Sat, 25 Jul 2026 23:03:31 -0700 Subject: [PATCH] Cycle every payload pattern per packet and always run both directions --- counters.go | 7 +++---- frame.go | 47 +++++++++++++++++++++---------------------- main.go | 58 +++++++++++++++++++---------------------------------- probe.go | 39 +++++++++++++++-------------------- rx.go | 2 +- ts.go | 15 ++++---------- tx.go | 9 +++++++-- ui.go | 2 -- 8 files changed, 75 insertions(+), 104 deletions(-) diff --git a/counters.go b/counters.go index 16555a1..b2c8b1f 100644 --- a/counters.go +++ b/counters.go @@ -6,10 +6,9 @@ import ( "strings" ) -// Everything the driver counts against a frame the link failed to carry, split -// by direction of travel so each counter belongs to exactly one direction: the -// transmitting interface owns the tx fields and the receiving interface the rx -// ones. Reading both sets off both interfaces would report every drop twice. +// 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", diff --git a/frame.go b/frame.go index 4696431..2ef9695 100644 --- a/frame.go +++ b/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) { diff --git a/main.go b/main.go index a811f2d..a0595df 100644 --- a/main.go +++ b/main.go @@ -414,15 +414,15 @@ func (d *direction) row(elapsed time.Duration, v view, target float64, length st } } -// Read once a second rather than per frame, since these are sysfs files; the -// display uses whatever the last sample left behind. +// 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) 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, @@ -434,7 +434,7 @@ func buildDirection(label string, tx, rx endpoint, patIdx int, sizes []int, cfg 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 { @@ -451,10 +451,9 @@ func buildDirection(label string, tx, rx endpoint, patIdx int, sizes []int, cfg 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.probeSpec = newFrameSpec(patIdx, rx.mac, tx.mac, cfg.probeEther, []int{probeSize}) + // 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) @@ -549,17 +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", 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") + 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, *zeroNS, *nsPerM); err != nil { + if err := run(*aName, *bName, *sizesArg, + *streams, *batch, *zeroNS, *nsPerM); err != nil { fmt.Fprintln(os.Stderr, "error:", err) os.Exit(1) } @@ -576,8 +573,8 @@ const ( totalsHold = 50 * time.Millisecond ) -func run(aName, bName, sizesArg, patArg string, - nStreams, batch int, duplex bool, zeroNS, nsPerM float64) 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") @@ -586,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 @@ -641,17 +634,12 @@ func run(aName, bName, sizesArg, patArg string, } 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 { @@ -678,23 +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"}, - {"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)", + {"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 diff --git a/probe.go b/probe.go index e522867..e252196 100644 --- a/probe.go +++ b/probe.go @@ -10,10 +10,10 @@ import ( ) 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 + // 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 @@ -27,10 +27,9 @@ const ( ) // 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. +// 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 @@ -52,10 +51,10 @@ func (v cableView) minText() string { return commasInt(v.min) } -// Summing both directions cancels the phy asymmetry between them, which is +// 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) != 2 { + if len(views) == 0 { return 0, false } var sum float64 @@ -65,7 +64,8 @@ func (c config) cableMetres(views []view) (float64, bool) { } sum += float64(v.cable.min) } - return (sum - c.zeroNS) / c.nsPerM, true + mean := sum / float64(len(views)) + return (mean - c.zeroNS) / c.nsPerM, true } func (c config) cableText(views []view) string { @@ -83,8 +83,6 @@ func newCableStats() *cableStats { } } -// 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() @@ -133,8 +131,6 @@ func (c *cableStats) reset() { 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 @@ -143,7 +139,7 @@ type probeSender struct { func (p *probeSender) run(done *atomic.Bool, startTx <-chan struct{}) { buf := make([]byte, probeSize) - p.spec.prefill(buf) + p.spec.prefill(buf, probePattern) oob := make([]byte, 512) scratch := make([]byte, 1) @@ -156,9 +152,8 @@ func (p *probeSender) run(done *atomic.Bool, startTx <-chan struct{}) { 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. + // 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 { @@ -166,8 +161,8 @@ func (p *probeSender) run(done *atomic.Bool, startTx <-chan struct{}) { } } - putHeader(buf, p.spec.patIdx, probeStream, seq, probeSize-minFrame, - p.spec.crcFor[probeSize]) + 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. @@ -211,8 +206,6 @@ func (p *probeSender) awaitTx(scratch, oob []byte) (int64, bool) { } } -// Reads probes on the far interface, where every frame carries a receive stamp -// from the MAC. type probeReceiver struct { fd int stats *cableStats diff --git a/rx.go b/rx.go index fa5b4c0..df97cd2 100644 --- a/rx.go +++ b/rx.go @@ -61,7 +61,7 @@ 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 } diff --git a/ts.go b/ts.go index c7f9cce..adf3a6e 100644 --- a/ts.go +++ b/ts.go @@ -12,17 +12,14 @@ const ( 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. +// 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 { @@ -70,9 +67,6 @@ func checkTimestamping(fd int, ifname string) checkResult { 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| @@ -85,9 +79,8 @@ func enableRxTimestamps(fd int) error { 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. +// 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) { diff --git a/tx.go b/tx.go index 4622a1b..292cc57 100644 --- a/tx.go +++ b/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) } diff --git a/ui.go b/ui.go index c4075ce..6fb47e7 100644 --- a/ui.go +++ b/ui.go @@ -289,8 +289,6 @@ func (d *display) render(dirs []*direction, views []view, elapsed time.Duration, fb.flush() } -// 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 != "" {