diff --git a/main.go b/main.go index c62db98..345a816 100644 --- a/main.go +++ b/main.go @@ -7,7 +7,6 @@ import ( "os" "os/signal" "slices" - "strconv" "strings" "sync" "sync/atomic" @@ -221,28 +220,6 @@ func lookupEndpoint(name string) (endpoint, error) { speed: float64(v) / 1000}, nil } -func parseSizes(s string) ([]int, error) { - var out []int - for _, f := range strings.Split(s, ",") { - f = strings.TrimSpace(f) - if f == "" { - continue - } - v, err := strconv.Atoi(f) - if err != nil { - return nil, fmt.Errorf("bad size %q: %w", f, err) - } - if v < minFrame { - return nil, fmt.Errorf("size %d below minimum %d", v, minFrame) - } - out = append(out, v) - } - if len(out) == 0 { - return nil, fmt.Errorf("no sizes given") - } - return out, nil -} - func (d *direction) snapshot() sample { var s sample for _, t := range d.txStats { @@ -447,10 +424,10 @@ func (d *direction) primeCounters() { d.prevConsole = d.base } -func buildDirection(label string, tx, rx endpoint, sizes []int, cfg config) (*direction, error) { +func buildDirection(label string, tx, rx endpoint) (*direction, error) { d := &direction{ short: tx.tag + "→" + rx.tag, - streams: newLossWindows(cfg.streams), + streams: newLossWindows(numStreams), cable: newCableStats(), } // Held open for the life of the run: the stats ioctl is issued five times a @@ -467,9 +444,9 @@ func buildDirection(label string, tx, rx endpoint, sizes []int, cfg config) (*di } d.win = newRateWindow(int(rateWindowSpan/sampleInterval) + 1) - for i := 0; i < cfg.streams; i++ { + for i := 0; i < numStreams; i++ { et := uint16(etherBase + i) - d.specs = append(d.specs, newFrameSpec(rx.mac, tx.mac, et, sizes)) + d.specs = append(d.specs, newFrameSpec(rx.mac, tx.mac, et, frameSizes)) fd, err := openTxSocket(tx.idx) if err != nil { @@ -488,7 +465,7 @@ func buildDirection(label string, tx, rx endpoint, sizes []int, cfg config) (*di // 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}) + d.probeSpec = newFrameSpec(rx.mac, tx.mac, probeEther, []int{probeSize}) fd, err := openTxSocket(tx.idx) if err != nil { return nil, fmt.Errorf("%s probe tx socket: %w", label, err) @@ -497,7 +474,7 @@ func buildDirection(label string, tx, rx endpoint, sizes []int, cfg config) (*di return nil, fmt.Errorf("%s probe tx timestamps: %w", label, err) } d.probeTxFD = fd - fd, err = openRxSocket(rx.idx, cfg.probeEther) + fd, err = openRxSocket(rx.idx, probeEther) if err != nil { return nil, fmt.Errorf("%s probe rx socket: %w", label, err) } @@ -509,13 +486,13 @@ func buildDirection(label string, tx, rx endpoint, sizes []int, cfg config) (*di return d, nil } -func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, cfg config, rxReady *sync.WaitGroup, startTx <-chan struct{}) { +func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, rxReady *sync.WaitGroup, startTx <-chan struct{}) { for i, fd := range d.txFDs { w := &txWorker{ fd: fd, stream: uint16(i), spec: d.specs[i], - batch: cfg.batch, + batch: batchSize, stats: d.txStats[i], startTx: startTx, } @@ -528,7 +505,7 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, cfg c for i, fd := range d.rxFDs { w := &rxWorker{ fd: fd, - batch: cfg.batch, + batch: batchSize, spec: d.specs[i], stats: d.rxStats[i], streams: d.streams, @@ -574,31 +551,26 @@ func (d *direction) close() { unix.Close(d.statFD) } -type config struct { - streams int - batch int - probeEther uint16 - nsPerM float64 -} +const ( + numStreams = 7 + batchSize = 64 + + probeEther uint16 = etherBase + numStreams +) + +var frameSizes = []int{64, 128, 256, 512, 1024, 1280, 1514} func main() { - var ( - // The names the kernel gives the only two ports built into it, since as - // PID 1 there is no udev to rename them and no command line to pass. - aName = flag.String("a", "eth0", "first interface") - bName = flag.String("b", "eth1", "second interface") - sizesArg = flag.String("sizes", "64,128,256,512,1024,1280,1514", "frame sizes in bytes, excluding FCS, cycled per packet") - 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") - - nsPerM = flag.Float64("ns-per-m", 4.85, "mean of both directions, per metre of cable") - ) + // The names the kernel gives the only two ports built into it, since as + // PID 1 there is no udev to rename them and no command line to pass. + aName := flag.String("a", "eth0", "first interface") + bName := flag.String("b", "eth1", "second interface") + nsPerM := flag.Float64("ns-per-m", 4.85, "mean of both directions, per metre of cable") flag.Parse() // Nothing here is recoverable by the time it reaches this point, and as PID 1 // a plain exit would panic the kernel anyway with less to show for it. - if err := run(*aName, *bName, *sizesArg, - *streams, *batch, *nsPerM); err != nil { + if err := run(*aName, *bName, *nsPerM); err != nil { panic(err) } } @@ -634,14 +606,7 @@ func (s *sampler) run(done *atomic.Bool, startTx <-chan struct{}) { } } -func run(aName, bName, sizesArg string, - nStreams, batch int, nsPerM float64) error { - - sizes, err := parseSizes(sizesArg) - if err != nil { - return err - } - +func run(aName, bName string, nsPerM float64) error { if err := reportChecks("BOOT", bootstrap()); err != nil { return err } @@ -655,7 +620,7 @@ func run(aName, bName, sizesArg string, return err } for _, e := range []endpoint{a, b} { - for _, s := range sizes { + for _, s := range frameSizes { if s > e.mtu+ethHdrLen { return fmt.Errorf("size %d exceeds %s MTU %d (max frame %d)", s, e.name, e.mtu, e.mtu+ethHdrLen) } @@ -665,10 +630,7 @@ func run(aName, bName, sizesArg string, a.tag, b.tag = "A", "B" ifnames := []string{a.name, b.name} - if nStreams < 1 { - return fmt.Errorf("need at least one stream") - } - ethertypes := make([]uint16, nStreams) + ethertypes := make([]uint16, numStreams) for i := range ethertypes { ethertypes[i] = uint16(etherBase + i) } @@ -677,16 +639,9 @@ func run(aName, bName, sizesArg string, return err } - cfg := config{ - streams: nStreams, - batch: batch, - probeEther: uint16(etherBase + nStreams), - nsPerM: nsPerM, - } - var dirs []*direction for _, p := range [][2]endpoint{{a, b}, {b, a}} { - d, err := buildDirection(p[0].name+"->"+p[1].name, p[0], p[1], sizes, cfg) + d, err := buildDirection(p[0].name+"->"+p[1].name, p[0], p[1]) if err != nil { return err } @@ -710,8 +665,8 @@ func run(aName, bName, sizesArg string, []bool{false, false, false, true, true}, linkRows)) target := a.speed - sizeStrs := make([]string, len(sizes)) - for i, s := range sizes { + sizeStrs := make([]string, len(frameSizes)) + for i, s := range frameSizes { sizeStrs[i] = fmt.Sprintf("%d", s) } fmt.Println(renderBox("CONFIG", @@ -719,9 +674,9 @@ func run(aName, bName, sizesArg string, []bool{false, false}, [][]string{ {"frame sizes", strings.Join(sizeStrs, " ")}, {"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)}, + numStreams, ethertypes[0], ethertypes[len(ethertypes)-1])}, + {"probe", fmt.Sprintf("ethertype 0x%04x every %s", probeEther, probeInterval)}, + {"batch", fmt.Sprintf("%d frames per syscall", batchSize)}, {"calibration", fmt.Sprintf("%g ns/m, zero taken from the shortest delay seen so far", nsPerM)}, {"buffers", fmt.Sprintf("sndbuf %s, rcvbuf %s", humanBytes(uint64(sockBufSize(dirs[0].txFDs[0], unix.SO_SNDBUF))), @@ -737,7 +692,7 @@ func run(aName, bName, sizesArg string, rxReady.Add(len(d.rxFDs) + 1) } for _, d := range dirs { - d.start(&wg, &doneTx, &doneRx, cfg, &rxReady, startTx) + d.start(&wg, &doneTx, &doneRx, &rxReady, startTx) } samp := &sampler{dirs: dirs} wg.Add(1) @@ -798,7 +753,7 @@ func run(aName, bName, sizesArg string, // Empty until the probe has a stamp from each direction, so the // panel shows nothing there rather than a placeholder. cable := "" - if m, ok := cfg.cableMetres(views); ok { + if m, ok := cableMetres(views, nsPerM); ok { cable = fmt.Sprintf("%.1f", m) } if err := disp.render(totalView(views), now.Sub(start), cable); err != nil { @@ -812,7 +767,7 @@ func run(aName, bName, sizesArg string, rows[i] = d.view(now) } length := "-" - if m, ok := cfg.cableMetres(rows); ok { + if m, ok := cableMetres(rows, nsPerM); ok { length = fmt.Sprintf("%.1f", m) } for i, d := range dirs { diff --git a/probe.go b/probe.go index 73c8c47..11711cf 100644 --- a/probe.go +++ b/probe.go @@ -54,7 +54,7 @@ func (v cableView) minText() string { // 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) { +func cableMetres(views []view, nsPerM float64) (float64, bool) { if len(views) == 0 { return 0, false } @@ -65,7 +65,7 @@ func (c config) cableMetres(views []view) (float64, bool) { } excess += float64(v.cable.min - v.cable.floor) } - return excess / float64(len(views)) / c.nsPerM, true + return excess / float64(len(views)) / nsPerM, true } func newCableStats() *cableStats {