Hardcode the settled sizes, streams and batch instead of carrying flags for them

This commit is contained in:
flamingcow
2026-08-04 13:31:14 -07:00
parent 05e1958710
commit 39c27166dc
2 changed files with 37 additions and 82 deletions
+33 -78
View File
@@ -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")
)
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 {
+2 -2
View File
@@ -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 {