From 7fa48da73c2c08b6a9a1be9f1d53934c0a18eefe Mon Sep 17 00:00:00 2001 From: flamingcow Date: Mon, 31 Aug 2026 14:33:09 -0700 Subject: [PATCH] Judge loss by arrival order and a deadline, one tx ring per stream --- direction.go | 15 +++- docs/measurement.md | 2 +- history.go | 9 +- loss.go | 140 +++++++++++------------------ loss_test.go | 211 +++++++++++++++----------------------------- main.go | 8 +- main_test.go | 8 +- phy.go | 4 + rx.go | 11 +-- sock.go | 7 +- system.go | 46 +++++++++- tx.go | 13 +++ 12 files changed, 220 insertions(+), 254 deletions(-) diff --git a/direction.go b/direction.go index 434eaf8..603cbd5 100644 --- a/direction.go +++ b/direction.go @@ -28,6 +28,8 @@ type direction struct { // Guards everything the sampler touches. Collection consumes self-clearing // sources, so a second concurrent reader would see a gap. + iface int + mu sync.Mutex prev sample nic *nicCounters @@ -214,8 +216,14 @@ type view struct { // One tick of every counter this direction owns, flattened to deltas: the // worker counters grow forever, the socket drop statistic clears itself on -// read, and the nic counters re-base, none of which survives past here. +// read, and the nic counters re-base, none of which survives past here. The +// loss judge runs first, so a silent wire's charges are on the books before +// the read. func (d *direction) collect() (frames, bytes uint64, e errs) { + now := time.Now() + for i := range d.streams { + d.streams[i].judge(now) + } cur := d.snapshot() prev := d.prev d.prev = cur @@ -308,7 +316,7 @@ func (d *direction) primeCounters() { d.mu.Unlock() } -func buildDirection(label string, tx, rx endpoint, masked *atomic.Bool) (*direction, error) { +func buildDirection(label string, iface int, tx, rx endpoint, masked *atomic.Bool) (*direction, error) { // Built before the windows, since each window judges sequence numbers against // the frontier its own sender publishes. txs := make([]*txStats, numStreams) @@ -316,6 +324,7 @@ func buildDirection(label string, tx, rx endpoint, masked *atomic.Bool) (*direct txs[i] = &txStats{} } d := &direction{ + iface: iface, txStats: txs, streams: newLossWindows(txs), masked: masked, @@ -360,6 +369,7 @@ func (d *direction) start(wg *sync.WaitGroup, done *atomic.Bool, rxReady *sync.W w := &txWorker{ fd: fd, stream: uint16(i), + cpu: txCPU(d.iface, i), spec: d.specs[i], batch: batchSize, stats: d.txStats[i], @@ -380,7 +390,6 @@ func (d *direction) start(wg *sync.WaitGroup, done *atomic.Bool, rxReady *sync.W spec: d.specs[i], stats: d.rxStats[i], loss: &d.streams[i], - masked: d.masked, ready: rxReady, } wg.Add(1) diff --git a/docs/measurement.md b/docs/measurement.md index c594871..2f56b85 100644 --- a/docs/measurement.md +++ b/docs/measurement.md @@ -69,7 +69,7 @@ Measured on the bench host (i9-13900H, 6P+8E, 20 threads), RAPL package power at Measured in the E810 era; the binding constraint was total CPU across ~28 goroutines on 20 threads. - **Splitting TX senders from RX streams** — raises TX but collapses RX (RX scales with queue count, capped at 7 on that NIC). -- **CPU pinning** — the Go scheduler beats manual placement; E-cores are poor at TX. +- **CPU pinning for performance** — the Go scheduler beats manual placement; E-cores are poor at TX. (The TX feeders are nevertheless each locked to a dedicated cpu, spaced off shared hyperthreads: XPS keys the hardware ring on the sending cpu, and a stable per-stream ring is an ordering requirement, not a speed one.) - **Batch sizes above 64** — no gain, worse loss. - **Chasing the TX `sendmmsg` busy-spin on ENOBUFS** — it only exists when the TX ring is full, which means the wire is the ceiling; recovering that CPU buys no packets. POLLOUT is inert under PACKET_QDISC_BYPASS (skb freed on ENOBUFS, socket always reports writable). - **Letting idle cores sleep by relaxing cpu_dma_latency** (200 µs admits the 127 µs C2_ACPI) — steady-state deep-idle residency ~2% and power unchanged: the Go scheduler spreads wakeups across every thread, so no core idles long enough for the idle governor to pick a deep state. diff --git a/history.go b/history.go index 2f82610..5b9c57c 100644 --- a/history.go +++ b/history.go @@ -16,9 +16,12 @@ const ( histSlotSpan = 5 * time.Second histSpan = histSlots * histSlotSpan - // Line rate is a verdict here, not a number: the headline is smeared and - // briefly gated at startup, so anything close to the target counts. - rateOKFrac = 0.98 + // Line rate is a verdict here, not a number: anything close to the target + // counts. Sub-millisecond host physics — an SMI, a DMA arbitration pause — + // shorts the odd display bucket by a few percent (94-97% bench-measured) + // and says nothing about the cable; a cable problem does not shave points, + // it craters the rate and brings link evidence with it. + rateOKFrac = 0.9 ) var faultClasses = []struct { diff --git a/loss.go b/loss.go index 967b211..e7275a8 100644 --- a/loss.go +++ b/loss.go @@ -1,22 +1,30 @@ package main import ( - "math/bits" "sync" "sync/atomic" + "time" ) -const ( - lossSlots = 1 << 16 - lossWords = lossSlots / 64 -) +// How long a sent frame may remain unreceived before it is charged as lost. +// The mask outlives a measure by the same window, so the judge's last +// masked-era charges still land inside it. +const lossDeadline = 100 * time.Millisecond +type frontier struct { + t time.Time + seq uint64 +} + +// Per-stream loss accounting with no tolerance of reordering: an arrival +// charges everything between the previous arrival and itself as lost, and +// judge charges whatever was sent more than lossDeadline ago without +// arriving. next is the first sequence number not yet accounted for, by +// arrival or by charge; anything arriving below it is late. type lossWindow struct { mu sync.Mutex - base uint64 - inited bool - maskedEra bool - bits []uint64 + next uint64 + frontiers []frontier lost atomic.Uint64 late atomic.Uint64 @@ -27,97 +35,49 @@ type lossWindow struct { } // Takes the sending halves rather than a count, so a window cannot be built -// without the bound it judges sequence numbers against. +// without the frontier it judges sequence numbers against. func newLossWindows(tx []*txStats) []lossWindow { w := make([]lossWindow, len(tx)) for i := range w { - w[i].bits = make([]uint64, lossWords) w[i].sent = &tx[i].sent } return w } -// A sequence number is only judged once it falls out of the window, so a frame -// that has arrived but not yet been drained is never miscounted as lost. -// -// Reports whether the sequence number could have come off the wire at all. One -// above what the sender has reached is a damaged header rather than a gap, and -// is refused before it can touch the base. Believing one would drag the window -// up past every real sequence, write off the span in between as lost, and leave -// every frame after it arriving below the base and counted late for the rest of -// the run. Nothing that arrives later is evidence enough to undo that, which is -// why the sender's own frontier is the guard rather than a limit on how far a -// single step may move. -// masked slides the window without charging. A gap's charge lands at eviction, -// which can fall after the mask lifts, so this source judges each charge -// against the mask itself rather than leaving a deferred judgment to the -// sampler's gate. -func (w *lossWindow) observe(seq uint64, masked bool) bool { - if seq >= w.sent.Load() { - return false - } +func (w *lossWindow) observe(seq uint64) { w.mu.Lock() - if w.maskedEra && !masked { - // Leaving a measure: the window was full when it began, so every hole - // still inside went missing while the mask was up. Marking them - // delivered keeps the write-off complete however the eviction timing - // falls, instead of charging the gap's in-window tail later. - for i := range w.bits { - w.bits[i] = ^uint64(0) - } - } - w.maskedEra = masked - if !w.inited { - // Start half a window below the first sequence seen, so anything the - // sender put on the wire before it lands inside the window rather than - // below the base. - if seq > lossSlots/2 { - w.base = seq - lossSlots/2 - } - w.inited = true - } - if seq < w.base { + if seq < w.next { w.mu.Unlock() - if !masked { - w.late.Add(1) - } - return true - } - if seq >= w.base+lossSlots { - w.evict(seq-lossSlots+1, masked) - } - idx := seq & (lossSlots - 1) - w.bits[idx>>6] |= 1 << (idx & 63) - w.mu.Unlock() - return true -} - -func (w *lossWindow) evict(newBase uint64, masked bool) { - span := newBase - w.base - if span >= lossSlots { - var missing uint64 - for i := range w.bits { - missing += uint64(64 - bits.OnesCount64(w.bits[i])) - w.bits[i] = 0 - } - if !masked { - w.lost.Add(missing + span - lossSlots) - } - w.base = newBase + w.late.Add(1) return } - var missing uint64 - for s := w.base; s < newBase; s++ { - idx := s & (lossSlots - 1) - word, bit := idx>>6, uint64(1)<<(idx&63) - if w.bits[word]&bit == 0 { - missing++ - continue - } - w.bits[word] &^= bit - } - if !masked { - w.lost.Add(missing) - } - w.base = newBase + lost := seq - w.next + w.next = seq + 1 + w.mu.Unlock() + w.lost.Add(lost) +} + +// Charges everything the sender had put on the wire by one deadline ago that +// has not arrived since, so a silent wire is charged within lossDeadline of +// going silent rather than when traffic resumes. +func (w *lossWindow) judge(now time.Time) { + w.mu.Lock() + var bound uint64 + keep := 0 + for i, f := range w.frontiers { + if now.Sub(f.t) < lossDeadline { + break + } + bound = f.seq + keep = i + 1 + } + w.frontiers = append(w.frontiers[:0], w.frontiers[keep:]...) + w.frontiers = append(w.frontiers, frontier{t: now, seq: w.sent.Load()}) + var lost uint64 + if bound > w.next { + lost = bound - w.next + w.next = bound + } + w.mu.Unlock() + w.lost.Add(lost) } diff --git a/loss_test.go b/loss_test.go index 3355aad..7f7c681 100644 --- a/loss_test.go +++ b/loss_test.go @@ -1,6 +1,9 @@ package main -import "testing" +import ( + "testing" + "time" +) // sent is the sender's frontier: sequence numbers at or above it were never put // on the wire. @@ -11,10 +14,10 @@ func newWindow(sent uint64) *lossWindow { return &w[0] } -func TestLossWindowContiguousLosesNothing(t *testing.T) { - w := newWindow(lossSlots + 1000) - for seq := uint64(0); seq < lossSlots+1000; seq++ { - w.observe(seq, false) +func TestLossContiguousLosesNothing(t *testing.T) { + w := newWindow(10000) + for seq := uint64(0); seq < 10000; seq++ { + w.observe(seq) } if got := w.lost.Load(); got != 0 { t.Errorf("lost = %d, want 0", got) @@ -24,149 +27,81 @@ func TestLossWindowContiguousLosesNothing(t *testing.T) { } } -func TestLossWindowCountsGapOnceEvicted(t *testing.T) { - w := newWindow(lossSlots + 1000) - for seq := uint64(0); seq < lossSlots+1000; seq++ { - if seq == 100 { - continue - } - w.observe(seq, false) +// An arrival is the judgment: the gap between it and the previous arrival is +// lost the moment it lands, not when some window slides. +func TestLossGapChargesOnArrival(t *testing.T) { + w := newWindow(10000) + for seq := uint64(0); seq < 100; seq++ { + w.observe(seq) } + w.observe(105) + if got := w.lost.Load(); got != 5 { + t.Errorf("lost = %d, want 5 the moment the gap's far edge arrived", got) + } +} + +// No tolerance of reordering: the newer frame writes the span off, and the +// older one arriving after it is late, not a repair. +func TestLossReorderChargesAndCountsLate(t *testing.T) { + w := newWindow(10000) + for seq := uint64(0); seq < 6; seq++ { + w.observe(seq) + } + w.observe(7) if got := w.lost.Load(); got != 1 { - t.Errorf("lost = %d, want 1", got) + t.Errorf("lost = %d after the skip, want 1", got) } -} - -// Arriving out of order inside the window is not loss: a sequence is only -// judged once it falls out the far end. -func TestLossWindowOutOfOrderIsNotLoss(t *testing.T) { - w := newWindow(lossSlots + 1000) - for seq := uint64(99); ; seq-- { - w.observe(seq, false) - if seq == 0 { - break - } - } - for seq := uint64(100); seq < lossSlots+1000; seq++ { - w.observe(seq, false) - } - if got := w.lost.Load(); got != 0 { - t.Errorf("lost = %d, want 0", got) - } -} - -// The other branch of evict: a jump past a whole window writes off everything -// the window held plus the sequences that never landed in it at all. -func TestLossWindowJumpBeyondWindow(t *testing.T) { - w := newWindow(200001) - w.observe(0, false) - w.observe(200000, false) - - // Everything below the new base except the one sequence that was seen. - want := uint64(200000 - lossSlots + 1 - 1) - if got := w.lost.Load(); got != want { - t.Errorf("lost = %d, want %d", got, want) - } - if got := w.late.Load(); got != 0 { - t.Errorf("late = %d, want 0", got) - } -} - -func TestLossWindowBelowBaseIsLate(t *testing.T) { - w := newWindow(100001) - w.observe(100000, false) - w.observe(1000, false) + w.observe(6) if got := w.late.Load(); got != 1 { t.Errorf("late = %d, want 1", got) } - if got := w.lost.Load(); got != 0 { - t.Errorf("lost = %d, want 0", got) - } -} - -// The first sequence seen starts the window half a span below it, so frames -// another worker is still holding land inside rather than arriving late. -func TestLossWindowStartsHalfAWindowBack(t *testing.T) { - w := newWindow(100001) - w.observe(100000, false) - if w.base != 100000-lossSlots/2 { - t.Errorf("base = %d, want %d", w.base, 100000-lossSlots/2) - } -} - -// The failure this guard exists for. A sequence number the sender never reached -// used to drag the base up to it, write off the span in between as lost, and -// leave every real frame after it below the base and counted late for the rest -// of the run. It has to be refused outright, and the stream has to go on -// counting as if it had never arrived. -func TestLossWindowRefusesUnsentSeq(t *testing.T) { - const first = 100000 - w := newWindow(first + 2000) - for seq := uint64(first); seq < first+1000; seq++ { - w.observe(seq, false) - } - base := w.base - - if w.observe(1<<62, false) { - t.Error("a sequence number far past the sender's frontier was accepted") - } - if got := w.lost.Load(); got != 0 { - t.Errorf("lost = %d after one impossible sequence number, want 0", got) - } - if w.base != base { - t.Errorf("base moved to %d, want it left at %d where the real traffic put it", - w.base, base) - } - - // Still tracking the real traffic, rather than reporting every frame late - // against a base that ran away. - for seq := uint64(first + 1000); seq < first+2000; seq++ { - w.observe(seq, false) - } - if got := w.late.Load(); got != 0 { - t.Errorf("late = %d, want 0", got) - } - if got := w.lost.Load(); got != 0 { - t.Errorf("lost = %d, want 0", got) - } -} - -// A gap written off while masked — a cable measure's own link blip — charges -// nothing, however the eviction timing falls: even the gap's tail still -// inside the window when the measure ends is presumed delivered. Counting -// resumes seamlessly and later real gaps are still caught. -func TestLossWindowMaskedSlidesWithoutCharging(t *testing.T) { - w := newWindow(4 * lossSlots) - for seq := uint64(0); seq < 100; seq++ { - w.observe(seq, false) - } - // The blip: a huge jump arriving during the measure. - w.observe(2*lossSlots, true) - if got := w.lost.Load(); got != 0 { - t.Errorf("lost = %d after a masked write-off, want 0", got) - } - - // The measure ends immediately — the worst case, with the gap's tail - // still in-window — and a single real gap follows. - for seq := uint64(2*lossSlots + 1); seq < 4*lossSlots-100; seq++ { - if seq == 2*lossSlots+500 { - continue - } - w.observe(seq, false) - } if got := w.lost.Load(); got != 1 { - t.Errorf("lost = %d, want exactly the one real post-measure gap", got) + t.Errorf("lost = %d after the late arrival, want still 1", got) } } -// The sender's own frontier is the bound, so the sequence one past it is -// refused while the one below it is not. -func TestLossWindowBoundIsExclusive(t *testing.T) { - w := newWindow(500) - if !w.observe(499, false) { - t.Error("the last sequence the sender put on the wire was refused") +// A silent wire is charged by the judge within one deadline: everything the +// frontier covered a deadline ago and nothing has delivered since is lost. +func TestLossJudgeChargesSilentWire(t *testing.T) { + w := newWindow(1000) + t0 := time.Now() + w.judge(t0) + w.judge(t0.Add(lossDeadline / 2)) + if got := w.lost.Load(); got != 0 { + t.Errorf("lost = %d before the deadline, want 0", got) } - if w.observe(500, false) { - t.Error("a sequence the sender had not reached was accepted") + w.judge(t0.Add(lossDeadline)) + if got := w.lost.Load(); got != 1000 { + t.Errorf("lost = %d after the deadline, want the whole silent frontier, 1000", got) + } +} + +// Arrivals move next past the old frontier, so the judge has nothing to add. +func TestLossJudgeSkipsDelivered(t *testing.T) { + w := newWindow(1000) + t0 := time.Now() + w.judge(t0) + for seq := uint64(0); seq < 1000; seq++ { + w.observe(seq) + } + w.judge(t0.Add(lossDeadline)) + if got := w.lost.Load(); got != 0 { + t.Errorf("lost = %d, want 0", got) + } +} + +// A frame the judge already wrote off arriving afterwards is late, and the +// charge is not repaired. +func TestLossJudgedFrameArrivingIsLate(t *testing.T) { + w := newWindow(1000) + t0 := time.Now() + w.judge(t0) + w.judge(t0.Add(lossDeadline)) + w.observe(500) + if got := w.late.Load(); got != 1 { + t.Errorf("late = %d, want 1", got) + } + if got := w.lost.Load(); got != 1000 { + t.Errorf("lost = %d, want unchanged 1000", got) } } diff --git a/main.go b/main.go index b32c3cd..437c729 100644 --- a/main.go +++ b/main.go @@ -104,6 +104,10 @@ func totalRow(elapsed time.Duration, v view, target float64, phy phyDisplay) []s } } +// One dedicated cpu per tx feeder, spaced two apart so no two spinners share +// a hyperthread pair; xps maps each cpu to its stream's own ring. +func txCPU(iface, stream int) int { return (iface*numStreams + stream) * 2 } + const ( numStreams = 2 batchSize = 64 @@ -242,8 +246,8 @@ func run(aName, bName string) (err error) { diag := newCableDiag(modules, cableInfo{}) 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], &diag.masked) + for i, p := range [][2]endpoint{{a, b}, {b, a}} { + d, err := buildDirection(p[0].name+"->"+p[1].name, i, p[0], p[1], &diag.masked) if err != nil { return err } diff --git a/main_test.go b/main_test.go index 3777b22..2349315 100644 --- a/main_test.go +++ b/main_test.go @@ -44,7 +44,7 @@ func TestRxBucketWraps(t *testing.T) { // complete across the board and the rate comes from the window behind the // lowest frontier. func TestDirectionRateReadsOneBucketBack(t *testing.T) { - d := &direction{rxStats: []*rxStats{{}, {}}} + d := &direction{masked: &atomic.Bool{}, rxStats: []*rxStats{{}, {}}} d.rxStats[0].commit(1, 1, 500) d.rxStats[1].commit(1, 1, 700) d.rxStats[0].commit(2, 1, 900) @@ -61,7 +61,7 @@ func TestDirectionRateReadsOneBucketBack(t *testing.T) { // stream still short of the epoch holds the read back rather than having its // half-filled bucket summed. func TestDirectionRateWaitsForSlowestStream(t *testing.T) { - d := &direction{rxStats: []*rxStats{{}, {}}} + d := &direction{masked: &atomic.Bool{}, rxStats: []*rxStats{{}, {}}} d.rxStats[0].commit(1, 1, 500) d.rxStats[0].commit(2, 1, 900) @@ -95,7 +95,7 @@ func TestDirectionRateWaitsForSlowestStream(t *testing.T) { // An epoch only advances when a frame is read, so a frontier that stops moving // is a quiet wire and must not keep reporting the last bucket. func TestDirectionRateGoesStale(t *testing.T) { - d := &direction{rxStats: []*rxStats{{}}} + d := &direction{masked: &atomic.Bool{}, rxStats: []*rxStats{{}}} d.rxStats[0].commit(1, 1, 500) d.rxStats[0].commit(2, 1, 900) @@ -123,7 +123,7 @@ func settleAll(frames, bytes []uint64) { // the raw ring per sample and displayed the excess again — 2C here, and >20G // on the wire. func TestDirectionRateNeverRedisplaysDonatedExcess(t *testing.T) { - d := &direction{rxStats: []*rxStats{{}}} + d := &direction{masked: &atomic.Bool{}, rxStats: []*rxStats{{}}} c := bucketWireCap seq := []uint64{c, 0, 2 * c, c, c, c, c, c} now := time.Now() diff --git a/phy.go b/phy.go index fae516d..b9eff5d 100644 --- a/phy.go +++ b/phy.go @@ -535,6 +535,10 @@ func (c *cableDiag) kick(done *atomic.Bool) bool { } } c.mu.Unlock() + // The judge can charge a masked-era loss up to one deadline after the + // wire went quiet, so the mask outlives the measure by that window — + // universally, rather than each source tracking its own tail. + time.Sleep(lossDeadline) c.masked.Store(false) }() return true diff --git a/rx.go b/rx.go index dc26c45..7176cb2 100644 --- a/rx.go +++ b/rx.go @@ -116,7 +116,6 @@ type rxWorker struct { spec *frameSpec stats *rxStats loss *lossWindow - masked *atomic.Bool ready *sync.WaitGroup } @@ -133,10 +132,6 @@ func (w *rxWorker) run(done *atomic.Bool) { w.ready.Done() for !done.Load() { - // The worker just counts; the mask gate above decides what a masked - // tick's counts are worth. One flag load covers the batch for the loss - // window, whose gap charges outlive the tick they happened in. - masked := w.masked.Load() n, err := recvmmsg(w.fd, hdrs, unix.MSG_WAITFORONE) if n <= 0 { if err != nil && err != unix.EAGAIN && err != unix.EINTR { @@ -161,12 +156,12 @@ func (w *rxWorker) run(done *atomic.Bool) { bytes += uint64(len(buf)) // The ethertype this socket is bound to already says which stream the - // frame belongs to, so a header naming another one is damaged, as is a - // sequence number the sender never reached. - if p.stream != w.stream || !w.loss.observe(p.seq, masked) { + // frame belongs to, so a header naming another one is damaged. + if p.stream != w.stream { w.stats.badHdr.Add(1) continue } + w.loss.observe(p.seq) want, ok := w.spec.expectedCRC(p.patIdx, p.payLen) if !ok { diff --git a/sock.go b/sock.go index 56fe1d4..1bfb325 100644 --- a/sock.go +++ b/sock.go @@ -16,9 +16,12 @@ type mmsghdr struct { func htons(v uint16) uint16 { return v<<8 | v>>8 } // Set through the FORCE options alone, since the plain ones are clamped to -// wmem_max and rmem_max and would quietly leave a fraction of this. +// wmem_max and rmem_max and would quietly leave a fraction of this. The send +// buffer is sized past the tx ring's worth of in-flight truesize, so the ring +// is what bounds the queue and a feeder pause has the full ring's slack +// before the wire runs dry. const ( - sndbufBytes = 8 << 20 + sndbufBytes = 64 << 20 rcvbufBytes = 64 << 20 ) diff --git a/system.go b/system.go index df46544..a836101 100644 --- a/system.go +++ b/system.go @@ -722,6 +722,45 @@ func checkRings(fd int, ifname string, rxWant, txWant uint32) checkResult { return res } +// With PACKET_QDISC_BYPASS the hardware ring comes from XPS, which maps the +// sending cpu to a ring, so an unpinned sender migrating across cpus spreads +// one stream over rings that drain concurrently and reorder it on the wire. +// Each stream's pinned feeder cpu maps to the stream's own ring instead. The +// rings sit past the flow-rule queues because ixgbe pairs rx and tx of one +// index on one interrupt vector, and tx cleanup does not belong on the +// vectors delivering the streams. +func checkTxQueues(ifname string, iface int) checkResult { + res := checkResult{item: ifname + " tx queues"} + queues, err := filepath.Glob("/sys/class/net/" + ifname + "/queues/tx-*/xps_cpus") + if err != nil || len(queues) == 0 { + res.err = fmt.Errorf("no tx queues found: %v", err) + return res + } + masks := make(map[string]string, numStreams) + for s := 0; s < numStreams; s++ { + masks[fmt.Sprintf("tx-%d", numStreams+s)] = fmt.Sprintf("%x", uint64(1)<