Give each stream its own ethertype steered to its own rx queue

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
+	patIdx    int
-	ref     []byte
+	ref       []byte
-	crcFor  map[int]uint32
+	crcFor    map[int]uint32
-	dstMAC  [6]byte
+	dstMAC    [6]byte
-	srcMAC  [6]byte
+	srcMAC    [6]byte
-	sizes   []int
+	etherType uint16
-	maxSize int
+	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,
+		patIdx:    patIdx,
-		ref:     ref,
+		ref:       ref,
-		crcFor:  crcFor,
+		crcFor:    crcFor,
-		dstMAC:  dst,
+		dstMAC:    dst,
-		srcMAC:  src,
+		srcMAC:    src,
-		sizes:   sizes,
+		etherType: etherType,
-		maxSize: maxSize,
+		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)
 }
 

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.streams),
-		streams: newLossWindows(cfg.txWorkers),
 		reports: make(chan string, 64),
 	}
 	d.nicTX = readNIC(tx.name)
 	d.nicRX = readNIC(rx.name)
 
-	fm, err := fanoutMode(cfg.fanout, cfg.rxWorkers)
+	for i := 0; i < cfg.streams; i++ {
-	if err != nil {
+		et := uint16(etherBase + i)
-		return nil, err
+		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, et)
-		fd, err := openRxSocket(rx.idx, cfg.fanoutID, fm)
 		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
+	streams int
-	rxWorkers int
+	batch   int
-	batch     int
+	txCPUs  []int
-	fanout    string
+	rxCPUs  []int
-	fanoutID  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")
+		streams  = flag.Int("streams", 4, "independent streams per direction; each gets its own ethertype, steered by a flow rule to its own rx queue")
-		rxN      = flag.Int("rx", 2, "rx workers per direction; more is slower, RSS cannot spread a raw ethertype across queues")
 		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,
+	if err := run(*aName, *bName, *sizesArg, *patArg, *txCPUs, *rxCPUs,
-		*txN, *rxN, *batch, *duplex); err != nil {
+		*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,
+func run(aName, bName, sizesArg, patArg, txCPUsArg, rxCPUsArg string,
-	txN, rxN, batch int, duplex bool) error {
+	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,
+		streams: nStreams,
-		rxWorkers: rxN,
+		batch:   batch,
-		batch:     batch,
+		txCPUs:  txCPUs,
-		fanout:    fanout,
+		rxCPUs:  rxCPUs,
-		txCPUs:    txCPUs,
-		rxCPUs:    rxCPUs,
 	}
 
 	var dirs []*direction
-	cfgA := cfg
+	d0, err := buildDirection(a.name+"->"+b.name, a, b, patIdx, sizes, cfg)
-	cfgA.fanoutID = 0x4341
-	d0, err := buildDirection(a.name+"->"+b.name, a, b, patIdx, sizes, cfgA)
 	if err != nil {
 		return err
 	}
 	dirs = append(dirs, d0)
 	if duplex {
-		cfgB := cfg
+		d1, err := buildDirection(b.name+"->"+a.name, b, a, patIdx, sizes, cfg)
-		cfgB.fanoutID = 0x4342
-		d1, err := buildDirection(b.name+"->"+a.name, b, a, patIdx, sizes, cfgB)
 		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)},
+			{"streams", fmt.Sprintf("%d per direction, ethertypes 0x%04x-0x%04x, one rx queue each",
-			{"workers", fmt.Sprintf("%d tx, %d rx per direction", txN, rxN)},
+				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))),

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
 }
 

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)