From 7e77d22f36665bf0e360e9dbb4673373ac057079 Mon Sep 17 00:00:00 2001 From: flamingcow Date: Sat, 25 Jul 2026 19:08:11 -0700 Subject: [PATCH] Give each stream its own ethertype steered to its own rx queue --- frame.go | 36 +++++------ main.go | 106 +++++++++++--------------------- sock.go | 9 +-- system.go | 177 +++++++++++++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 232 insertions(+), 96 deletions(-) diff --git a/frame.go b/frame.go index 3090c8f..007aaaa 100644 --- a/frame.go +++ b/frame.go @@ -8,7 +8,7 @@ import ( ) const ( - etherType = 0x88b5 + etherBase = 0x88b5 ethHdrLen = 14 hdrLen = 24 hdrMagic = 0x43424c54 @@ -75,16 +75,17 @@ func patternIndex(name string) (int, error) { } type frameSpec struct { - patIdx int - ref []byte - crcFor map[int]uint32 - dstMAC [6]byte - srcMAC [6]byte - sizes []int - maxSize int + patIdx int + ref []byte + crcFor map[int]uint32 + dstMAC [6]byte + srcMAC [6]byte + etherType uint16 + sizes []int + maxSize int } -func newFrameSpec(patIdx int, dst, src [6]byte, sizes []int) *frameSpec { +func newFrameSpec(patIdx int, dst, src [6]byte, etherType uint16, sizes []int) *frameSpec { maxSize := 0 for _, s := range sizes { if s > maxSize { @@ -98,20 +99,21 @@ func newFrameSpec(patIdx int, dst, src [6]byte, sizes []int) *frameSpec { crcFor[s] = crc32.Checksum(ref[:s-minFrame], crcTable) } return &frameSpec{ - patIdx: patIdx, - ref: ref, - crcFor: crcFor, - dstMAC: dst, - srcMAC: src, - sizes: sizes, - maxSize: maxSize, + patIdx: patIdx, + ref: ref, + crcFor: crcFor, + dstMAC: dst, + srcMAC: src, + etherType: etherType, + sizes: sizes, + maxSize: maxSize, } } func (f *frameSpec) prefill(buf []byte) { copy(buf[0:6], f.dstMAC[:]) copy(buf[6:12], f.srcMAC[:]) - binary.BigEndian.PutUint16(buf[12:14], etherType) + binary.BigEndian.PutUint16(buf[12:14], f.etherType) copy(buf[minFrame:], f.ref) } diff --git a/main.go b/main.go index 58f28aa..dab4f81 100644 --- a/main.go +++ b/main.go @@ -16,13 +16,6 @@ import ( "golang.org/x/sys/unix" ) -const ( - fanoutHash = 0 - fanoutLB = 1 - fanoutCPU = 2 - fanoutRollover = 3 -) - const wireOverhead = 24 type endpoint struct { @@ -44,7 +37,7 @@ type direction struct { short string tx endpoint rx endpoint - spec *frameSpec + specs []*frameSpec txStats []*txStats rxStats []*rxStats streams []lossWindow @@ -128,25 +121,6 @@ func cpuAt(cpus []int, i int) int { return cpus[i%len(cpus)] } -func fanoutMode(name string, nsock int) (int, error) { - if nsock < 2 { - return -1, nil - } - switch name { - case "none": - return -1, nil - case "hash": - return fanoutHash, nil - case "lb": - return fanoutLB, nil - case "cpu": - return fanoutCPU, nil - case "rollover": - return fanoutRollover, nil - } - return 0, fmt.Errorf("unknown fanout mode %q", name) -} - func (d *direction) snapshot() sample { var s sample for _, t := range d.txStats { @@ -244,30 +218,26 @@ func buildDirection(label string, tx, rx endpoint, patIdx int, sizes []int, cfg short: tx.tag + "→" + rx.tag, tx: tx, rx: rx, - spec: newFrameSpec(patIdx, rx.mac, tx.mac, sizes), - streams: newLossWindows(cfg.txWorkers), + streams: newLossWindows(cfg.streams), reports: make(chan string, 64), } d.nicTX = readNIC(tx.name) d.nicRX = readNIC(rx.name) - fm, err := fanoutMode(cfg.fanout, cfg.rxWorkers) - if err != nil { - return nil, err - } + for i := 0; i < cfg.streams; i++ { + et := uint16(etherBase + i) + d.specs = append(d.specs, newFrameSpec(patIdx, rx.mac, tx.mac, et, sizes)) - for i := 0; i < cfg.txWorkers; i++ { fd, err := openTxSocket(tx.idx) if err != nil { return nil, fmt.Errorf("%s tx socket: %w", label, err) } d.txFDs = append(d.txFDs, fd) d.txStats = append(d.txStats, &txStats{}) - } - for i := 0; i < cfg.rxWorkers; i++ { - fd, err := openRxSocket(rx.idx, cfg.fanoutID, fm) + + fd, err = openRxSocket(rx.idx, et) if err != nil { - return nil, fmt.Errorf("%s rx socket: %w", label, err) + return nil, fmt.Errorf("%s rx socket for 0x%04x: %w", label, et, err) } d.rxFDs = append(d.rxFDs, fd) d.rxStats = append(d.rxStats, &rxStats{}) @@ -280,7 +250,7 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, cfg c w := &txWorker{ fd: fd, stream: uint16(i), - spec: d.spec, + spec: d.specs[i], batch: cfg.batch, cpu: cpuAt(cfg.txCPUs, i), stats: d.txStats[i], @@ -297,7 +267,7 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, cfg c fd: fd, batch: cfg.batch, cpu: cpuAt(cfg.rxCPUs, i), - spec: d.spec, + spec: d.specs[i], stats: d.rxStats[i], streams: d.streams, reports: d.reports, @@ -321,13 +291,10 @@ func (d *direction) close() { } type config struct { - txWorkers int - rxWorkers int - batch int - fanout string - fanoutID int - txCPUs []int - rxCPUs []int + streams int + batch int + txCPUs []int + rxCPUs []int } func main() { @@ -336,18 +303,16 @@ func main() { bName = flag.String("b", "", "second interface") sizesArg = flag.String("sizes", "1024,1280,1514", "frame sizes in bytes, excluding FCS, cycled per packet; below ~900 the host cannot keep up and loss stops meaning anything") patArg = flag.String("pattern", "prbs", "payload pattern") - txN = flag.Int("tx", 1, "tx workers per direction; one saturates 10G at these sizes") - rxN = flag.Int("rx", 2, "rx workers per direction; more is slower, RSS cannot spread a raw ethertype across queues") + streams = flag.Int("streams", 4, "independent streams per direction; 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") - fanout = flag.String("fanout", "lb", "rx fanout mode: none, hash, lb, cpu, rollover") duplex = flag.Bool("duplex", true, "run both directions simultaneously") txCPUs = flag.String("txcpus", "", "comma-separated CPUs to pin tx workers to") rxCPUs = flag.String("rxcpus", "", "comma-separated CPUs to pin rx workers to") ) flag.Parse() - if err := run(*aName, *bName, *sizesArg, *patArg, *fanout, *txCPUs, *rxCPUs, - *txN, *rxN, *batch, *duplex); err != nil { + if err := run(*aName, *bName, *sizesArg, *patArg, *txCPUs, *rxCPUs, + *streams, *batch, *duplex); err != nil { fmt.Fprintln(os.Stderr, "error:", err) os.Exit(1) } @@ -355,8 +320,8 @@ func main() { const reportInterval = 500 * time.Millisecond -func run(aName, bName, sizesArg, patArg, fanout, txCPUsArg, rxCPUsArg string, - txN, rxN, batch int, duplex bool) error { +func run(aName, bName, sizesArg, patArg, txCPUsArg, rxCPUsArg string, + nStreams, batch int, duplex bool) error { if aName == "" || bName == "" { return fmt.Errorf("both -a and -b are required") @@ -396,9 +361,17 @@ func run(aName, bName, sizesArg, patArg, fanout, txCPUsArg, rxCPUsArg 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) + for i := range ethertypes { + ethertypes[i] = uint16(etherBase + i) + } + var fatal []string var tuneRows [][]string - for _, r := range configureSystem(ifnames) { + for _, r := range configureSystem(ifnames, ethertypes) { tuneRows = append(tuneRows, []string{r.item, r.status(), r.detail()}) if r.fatal { fatal = append(fatal, r.item) @@ -412,26 +385,20 @@ func run(aName, bName, sizesArg, patArg, fanout, txCPUsArg, rxCPUsArg string, } cfg := config{ - txWorkers: txN, - rxWorkers: rxN, - batch: batch, - fanout: fanout, - txCPUs: txCPUs, - rxCPUs: rxCPUs, + streams: nStreams, + batch: batch, + txCPUs: txCPUs, + rxCPUs: rxCPUs, } var dirs []*direction - cfgA := cfg - cfgA.fanoutID = 0x4341 - d0, err := buildDirection(a.name+"->"+b.name, a, b, patIdx, sizes, cfgA) + d0, err := buildDirection(a.name+"->"+b.name, a, b, patIdx, sizes, cfg) if err != nil { return err } dirs = append(dirs, d0) if duplex { - cfgB := cfg - cfgB.fanoutID = 0x4342 - d1, err := buildDirection(b.name+"->"+a.name, b, a, patIdx, sizes, cfgB) + d1, err := buildDirection(b.name+"->"+a.name, b, a, patIdx, sizes, cfg) if err != nil { return err } @@ -467,11 +434,10 @@ func run(aName, bName, sizesArg, patArg, fanout, txCPUsArg, rxCPUsArg string, []bool{false, false}, [][]string{ {"pattern", patterns[patIdx].name}, {"frame sizes", strings.Join(sizeStrs, " ")}, - {"ethertype", fmt.Sprintf("0x%04x", etherType)}, - {"workers", fmt.Sprintf("%d tx, %d rx per direction", txN, rxN)}, + {"streams", fmt.Sprintf("%d per direction, ethertypes 0x%04x-0x%04x, one rx queue each", + nStreams, ethertypes[0], ethertypes[len(ethertypes)-1])}, {"batch", fmt.Sprintf("%d frames per syscall", batch)}, {"payload verify", "crc32c on every frame"}, - {"rx fanout", fanout}, {"duplex", fmt.Sprintf("%v", duplex)}, {"socket buffers", fmt.Sprintf("sndbuf %s, rcvbuf %s (granted)", humanBytes(uint64(sockBufSize(dirs[0].txFDs[0], unix.SO_SNDBUF))), diff --git a/sock.go b/sock.go index ac99258..aee6ab1 100644 --- a/sock.go +++ b/sock.go @@ -56,7 +56,7 @@ func openTxSocket(ifindex int) (int, error) { return fd, nil } -func openRxSocket(ifindex, fanoutID, fanoutMode int) (int, error) { +func openRxSocket(ifindex int, etherType uint16) (int, error) { proto := int(htons(etherType)) fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, proto) if err != nil { @@ -82,13 +82,6 @@ func openRxSocket(ifindex, fanoutID, fanoutMode int) (int, error) { unix.Close(fd) return -1, fmt.Errorf("rcvtimeo: %w", err) } - if fanoutMode >= 0 { - arg := (fanoutMode << 16) | (fanoutID & 0xffff) - if err := unix.SetsockoptInt(fd, unix.SOL_PACKET, unix.PACKET_FANOUT, arg); err != nil { - unix.Close(fd) - return -1, fmt.Errorf("fanout: %w", err) - } - } return fd, nil } diff --git a/system.go b/system.go index 4750c6f..a865538 100644 --- a/system.go +++ b/system.go @@ -1,6 +1,7 @@ package main import ( + "encoding/binary" "fmt" "os" "path/filepath" @@ -11,6 +12,179 @@ import ( "golang.org/x/sys/unix" ) +const ( + ethtoolGRXRINGS = 0x2d + ethtoolGRXCLSRLCNT = 0x2e + ethtoolGRXCLSRULE = 0x2f + ethtoolGRXCLSRLALL = 0x30 + ethtoolSRXCLSRLDEL = 0x31 + ethtoolSRXCLSRLINS = 0x32 + etherFlow = 0x12 +) + +type ethtoolFlowExt struct { + padding [2]byte + hDest [6]byte + vlanEtype uint16 + vlanTci uint16 + data [2]uint32 +} + +type ethtoolRxFlowSpec struct { + flowType uint32 + hU [52]byte + hExt ethtoolFlowExt + mU [52]byte + mExt ethtoolFlowExt + _ [4]byte + ringCookie uint64 + location uint32 + _ [4]byte +} + +type ethtoolRxnfc struct { + cmd uint32 + flowType uint32 + data uint64 + fs ethtoolRxFlowSpec + ruleCnt uint32 + _ [4]byte +} + +func rxRings(fd int, ifname string) (uint64, error) { + nfc := ethtoolRxnfc{cmd: ethtoolGRXRINGS} + if err := ethtoolCall(fd, ifname, unsafe.Pointer(&nfc)); err != nil { + return 0, err + } + return nfc.data, nil +} + +// Returns the installed rule locations and the total filter capacity. ice +// only honours filters near the top of that range, which is why ethtool's own +// rule manager allocates downwards from the end. +func allRuleLocations(fd int, ifname string) ([]uint32, uint32, error) { + cnt := ethtoolRxnfc{cmd: ethtoolGRXCLSRLCNT} + if err := ethtoolCall(fd, ifname, unsafe.Pointer(&cnt)); err != nil { + return nil, 0, err + } + capacity := uint32(cnt.data) + if cnt.ruleCnt == 0 { + return nil, capacity, nil + } + + // rule_locs follows rule_cnt directly, before the struct's tail padding, + // and that is where the kernel copies it to. + locOff := unsafe.Offsetof(ethtoolRxnfc{}.ruleCnt) + 4 + buf := make([]byte, int(unsafe.Sizeof(ethtoolRxnfc{}))+4*int(cnt.ruleCnt)) + all := (*ethtoolRxnfc)(unsafe.Pointer(&buf[0])) + all.cmd = ethtoolGRXCLSRLALL + all.ruleCnt = cnt.ruleCnt + if err := ethtoolCall(fd, ifname, unsafe.Pointer(&buf[0])); err != nil { + return nil, 0, err + } + + raw := buf[locOff:] + locs := make([]uint32, 0, all.ruleCnt) + for i := 0; i < int(all.ruleCnt); i++ { + locs = append(locs, binary.LittleEndian.Uint32(raw[i*4:])) + } + return locs, capacity, nil +} + +func ruleIsEther(fd int, ifname string, loc uint32) bool { + get := ethtoolRxnfc{cmd: ethtoolGRXCLSRULE} + get.fs.location = loc + if err := ethtoolCall(fd, ifname, unsafe.Pointer(&get)); err != nil { + return false + } + return get.fs.flowType&0xff == etherFlow +} + +func deleteRule(fd int, ifname string, loc uint32) error { + nfc := ethtoolRxnfc{cmd: ethtoolSRXCLSRLDEL} + nfc.fs.location = loc + return ethtoolCall(fd, ifname, unsafe.Pointer(&nfc)) +} + +// ice rejects RX_CLS_LOC_ANY, so the caller must supply a free location. +func insertEtherRule(fd int, ifname string, ethType uint16, queue uint64, loc uint32) error { + nfc := ethtoolRxnfc{cmd: ethtoolSRXCLSRLINS} + nfc.fs.flowType = etherFlow + binary.BigEndian.PutUint16(nfc.fs.hU[12:14], ethType) + // A set mask bit means that bit must match, so mask only the ethertype and + // leave both MAC masks zero. Note ethtool -n prints the complement of this. + binary.BigEndian.PutUint16(nfc.fs.mU[12:14], 0xffff) + nfc.fs.ringCookie = queue + nfc.fs.location = loc + return ethtoolCall(fd, ifname, unsafe.Pointer(&nfc)) +} + +func checkFlowRules(fd int, ifname string, ethertypes []uint16) checkResult { + res := checkResult{item: ifname + " flow rules"} + rings, err := rxRings(fd, ifname) + if err != nil { + res.err = err + res.fatal = true + return res + } + if uint64(len(ethertypes)) > rings { + res.err = fmt.Errorf("%d streams needs %d rx rings, only %d available", + len(ethertypes), len(ethertypes), rings) + res.fatal = true + return res + } + + locs, capacity, err := allRuleLocations(fd, ifname) + if err != nil { + res.err = err + res.fatal = true + return res + } + if capacity < uint32(len(ethertypes)) { + res.err = fmt.Errorf("filter capacity %d is below %d streams", capacity, len(ethertypes)) + res.fatal = true + return res + } + var stale []uint32 + taken := make(map[uint32]bool, len(locs)) + for _, loc := range locs { + if ruleIsEther(fd, ifname, loc) { + stale = append(stale, loc) + continue + } + taken[loc] = true + } + for _, loc := range stale { + if err := deleteRule(fd, ifname, loc); err != nil { + res.err = fmt.Errorf("deleting stale rule %d: %w", loc, err) + res.fatal = true + return res + } + } + + next := capacity - 1 + for i, et := range ethertypes { + for taken[next] && next > 0 { + next-- + } + if err := insertEtherRule(fd, ifname, et, uint64(i), next); err != nil { + res.err = fmt.Errorf("steering ethertype 0x%04x to queue %d at location %d: %w", + et, i, next, err) + res.fatal = true + return res + } + taken[next] = true + } + + res.fixed = len(stale) > 0 + res.state = fmt.Sprintf("0x%04x-0x%04x to queues 0-%d of %d", + ethertypes[0], ethertypes[len(ethertypes)-1], len(ethertypes)-1, rings) + if len(stale) > 0 { + res.state = fmt.Sprintf("replaced %d stale, %s", len(stale), res.state) + } + return res +} + type ethtoolIfreq struct { name [unix.IFNAMSIZ]byte data unsafe.Pointer @@ -280,12 +454,13 @@ func withIoctlSocket(fn func(fd int) []checkResult) []checkResult { return fn(fd) } -func configureSystem(ifnames []string) []checkResult { +func configureSystem(ifnames []string, ethertypes []uint16) []checkResult { return withIoctlSocket(func(fd int) []checkResult { out := []checkResult{checkGovernor(wantGovernor)} for _, ifname := range ifnames { out = append(out, checkLinkUp(fd, ifname)) out = append(out, checkCoalesce(fd, ifname, wantCoalesceUsecs, wantCoalesceUsecs)) + out = append(out, checkFlowRules(fd, ifname, ethertypes)) carrierWait := 3 * time.Second r, reset := checkRings(fd, ifname, wantRxRing, wantTxRing)