Default to seven steered streams across the full frame size range

main.go

@@ -301,9 +301,9 @@ func main() {
 	var (
 		aName    = flag.String("a", "", "first interface")
 		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")
+		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", 4, "independent streams per direction; each gets its own ethertype, steered by a flow rule to its own rx queue")
+		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")
 		txCPUs   = flag.String("txcpus", "", "comma-separated CPUs to pin tx workers to")

system.go

@@ -119,6 +119,33 @@ func insertEtherRule(fd int, ifname string, ethType uint16, queue uint64, loc ui
 	return ethtoolCall(fd, ifname, unsafe.Pointer(&nfc))
 }
 
+// Our rules survive process exit, so stale ones are cleared before the queue
+// configuration is touched and fresh ones installed.
+func clearFlowRules(fd int, ifname string) checkResult {
+	res := checkResult{item: ifname + " stale rules"}
+	locs, _, err := allRuleLocations(fd, ifname)
+	if err != nil {
+		res.err = err
+		res.fatal = true
+		return res
+	}
+	n := 0
+	for _, loc := range locs {
+		if !ruleIsEther(fd, ifname, loc) {
+			continue
+		}
+		if err := deleteRule(fd, ifname, loc); err != nil {
+			res.err = fmt.Errorf("deleting rule %d: %w", loc, err)
+			res.fatal = true
+			return res
+		}
+		n++
+	}
+	res.fixed = n > 0
+	res.state = fmt.Sprintf("%d removed", n)
+	return res
+}
+
 func checkFlowRules(fd int, ifname string, ethertypes []uint16) checkResult {
 	res := checkResult{item: ifname + " flow rules"}
 	rings, err := rxRings(fd, ifname)
@@ -145,22 +172,10 @@ func checkFlowRules(fd int, ifname string, ethertypes []uint16) checkResult {
 		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 {
@@ -176,12 +191,8 @@ func checkFlowRules(fd int, ifname string, ethertypes []uint16) checkResult {
 		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
 }
 
@@ -459,16 +470,19 @@ func configureSystem(ifnames []string, ethertypes []uint16) []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, clearFlowRules(fd, ifname))
-			out = append(out, checkFlowRules(fd, ifname, ethertypes))
 
+			// Ring changes reprogram the queues, so flow rules pointing at those
+			// queues have to be installed afterwards.
 			carrierWait := 3 * time.Second
-			r, reset := checkRings(fd, ifname, wantRxRing, wantTxRing)
+			ringRes, ringReset := checkRings(fd, ifname, wantRxRing, wantTxRing)
-			out = append(out, r)
+			out = append(out, ringRes)
-			if reset {
+			if ringReset {
 				carrierWait = 10 * time.Second
 			}
 			out = append(out, checkCarrier(ifname, carrierWait))
+			out = append(out, checkCoalesce(fd, ifname, wantCoalesceUsecs, wantCoalesceUsecs))
+			out = append(out, checkFlowRules(fd, ifname, ethertypes))
 		}
 		return out
 	})