tun: Hold one transmit mode per routine

With staged transmit, an idle reader encrypted and sent a small read
itself, while larger reads went to the workers and left from the
ordered sender. Each read chose on its own, so under load a routine
alternated between the two paths. Their packets left from different
CPUs, and the receiver's TCP saw them out of order: on 100 GbE, summed
over four hosts, dsack_dups rose from 0-1 on master to about 25k and
reord_seen from 0 to 13k-19k.

Each routine now holds one mode, inline on the reader or fanout to the
workers and the sender. It switches only after tun.staged.gate_hold
(default 64) reads in a row want the other mode, and before going
inline it waits until every batch in flight is on the wire, so a
switch never lets a later packet overtake an earlier one. Going to
fanout nothing is in flight. flow.tx.gate_switches counts the switches.
On the same hosts dsack_dups fell to 20-52 and reord_seen to 0-2,
while throughput matched or beat per-read mixing (many-to-one 34.9
instead of 30.7 Gbit/s).

The affinity test waited only for the previous batch's packets to be
captured before the next read, but the sender returns the slot after
its write, so the reader could still see it in flight and stage the
small read. Switching to inline now drains, which makes the inline
send deterministic there.

examples/config.yml

@@ -284,9 +284,14 @@ tun:
   # Staged reservations may use half the window, the other half absorbs inline and control traffic, relay forwarding
   # and network reordering; unlimited forwarding or delay can still exceed it. Waiting for credit on a busy tunnel
   # also delays the other tunnels of the same read, and readers are not guaranteed fairness.
-  # Idle batches below inline_below TX segments or RX packets stay on the reader, default 256. With multiple routines,
-  # those sends still take tunnel reservation credit. 0 hands every batch to the workers and the ordered sender or
-  # writer; only TX reads too large for a tunnel's whole credit are still sent by the reader, after it drains.
+  # Small batches, below inline_below TX segments or RX packets (default 256), can skip the workers. On transmit each
+  # routine holds one mode, inline on the reader or fanout to the workers and the ordered sender, and switches only
+  # after gate_hold (default 64) reads in a row want the other one. Before going inline it waits for every batch in
+  # flight to reach the wire, so later packets never overtake earlier ones; flow.tx.gate_switches counts the switches.
+  # Choosing per read instead let small inline sends leave from another CPU than staged ones, and TCP saw reordering.
+  # On receive a listener with nothing in flight writes a small batch itself. With multiple routines, inline sends
+  # still take tunnel reservation credit. 0 hands every batch to the workers and the ordered sender or writer; only TX
+  # reads too large for a tunnel's whole credit are still sent by the reader, after it drains.
   # tx: false keeps transmit inline and stages only receive.
   # tx_slots and rx_slots bound the batches each routine has in flight per direction, minimum 2. A TX slot holds one
   # tun read and its ciphertext, about 1.1 MiB; the default is 8, or workers + 2 if more. An RX slot holds one
@@ -296,6 +301,7 @@ tun:
   #  workers: 0
   #  tx: true
   #  inline_below: 256
+  #  gate_hold: 64
   #  tx_slots: 8
   #  rx_slots: 6
 

flow.go

@@ -23,10 +23,11 @@ import (
 // tun.staged.tx_slots and rx_slots override the counts. Small-MSS TX arenas can grow.
 //
 // Metrics are registered only for active pipelines. flow.{tx,rx}.batches counts reads handed to the ordered stage and
-// flow.{tx,rx}.inline those an idle reader handled itself (TX inline also counts oversized reads); their rates show
-// whether staging is used. flow.stage.<stage>.q<queue>.busy_ns locates the limiting stage: tx_read, tx_encrypt,
-// tx_send, rx_read, rx_decrypt or rx_write. Take rates; 1e9 ns/s fills a serial stage. These are elapsed times that
-// exclude read syscalls and downstream waits; descheduling inflates them and worker times add up across the pool.
+// flow.{tx,rx}.inline those the reader handled itself (TX inline also counts oversized reads); their rates show
+// whether staging is used. flow.tx.gate_switches counts transmit switches between inline and fanout mode.
+// flow.stage.<stage>.q<queue>.busy_ns locates the limiting stage: tx_read, tx_encrypt, tx_send, rx_read, rx_decrypt or
+// rx_write. Take rates; 1e9 ns/s fills a serial stage. These are elapsed times that exclude read syscalls and
+// downstream waits; descheduling inflates them and worker times add up across the pool.
 // Inline work counts at the reader; small RX batches sent to the writer include decryption in writer time.
 
 // stagedConfig is tun.staged. workers is per routine and direction; zero keeps processing inline.
@@ -34,12 +35,19 @@ type stagedConfig struct {
 	workers int
 	tx      bool
 	rx      bool
-	// inlineBelow is the TX segment or RX packet count below which an idle reader skips the worker handoff, 0 never
+	// inlineBelow is the TX segment or RX packet count below which a read may skip the worker handoff, 0 never: a TX
+	// read in inline mode, an RX read with nothing in flight
 	inlineBelow int
 	// txSlots and rxSlots are how many batches each routine can have in flight per direction
 	txSlots, rxSlots int
+	// gateHold is how many TX reads in a row must want the other mode before a routine switches between inline and
+	// fanout transmit
+	gateHold int
 }
 
+// stagedGateHold is the default tun.staged.gate_hold
+const stagedGateHold = 64
+
 const (
 	// stagedTxSlots is the default TX slot count, unless the workers need more
 	stagedTxSlots = 8
@@ -85,10 +93,12 @@ func parseStagedConfig(c *config.C, l *slog.Logger) stagedConfig {
 		l.Warn("tun.staged cannot encrypt in parallel with boringcrypto or fips140; transmit stays inline")
 		sc.tx = false
 	}
+	sc.gateHold = max(c.GetInt("tun.staged.gate_hold", stagedGateHold), 1)
 	sc.txSlots, sc.rxSlots = stagedSlots(sc.workers, c.GetInt("listen.batch", 64),
 		c.GetInt("tun.staged.tx_slots", 0), c.GetInt("tun.staged.rx_slots", 0))
 	l.Info("Staged packet processing enabled", "workers", sc.workers, "tx", sc.tx, "rx", sc.rx,
-		"inlineBelow", sc.inlineBelow, "txSlots", sc.txSlots, "rxSlots", sc.rxSlots)
+		"inlineBelow", sc.inlineBelow, "gateHold", sc.gateHold,
+		"txSlots", sc.txSlots, "rxSlots", sc.rxSlots)
 	return sc
 }
 

flow_gate_test.go

@@ -0,0 +1,135 @@
+//go:build linux && !android && !e2e_testing
+
+package nebula
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/flynn/noise"
+	"github.com/rcrowley/go-metrics"
+	"github.com/slackhq/nebula/overlay/tio"
+	"github.com/slackhq/nebula/udp"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+)
+
+// gated is staging with an inline threshold of 4 segments and the given gate hold.
+func gated(hold int) stagedConfig {
+	sc := adaptive(2)
+	sc.inlineBelow, sc.gateHold = 4, hold
+	return sc
+}
+
+// gateCounters snapshots the transmit counters a gate test checks.
+type gateCounters struct{ inline, batches, switches int64 }
+
+func readGateCounters() gateCounters {
+	return gateCounters{
+		inline:   metrics.GetOrRegisterCounter("flow.tx.inline", nil).Count(),
+		batches:  metrics.GetOrRegisterCounter("flow.tx.batches", nil).Count(),
+		switches: metrics.GetOrRegisterCounter("flow.tx.gate_switches", nil).Count(),
+	}
+}
+
+func (c gateCounters) since(c0 gateCounters) gateCounters {
+	return gateCounters{c.inline - c0.inline, c.batches - c0.batches, c.switches - c0.switches}
+}
+
+// gateReads builds reads of small (3 segments, below the threshold) or large (2 superpackets of 9) shape, in order,
+// for one flow, and the segments they carry.
+func gateReads(large ...bool) ([][]tio.Packet, []decodedSeg) {
+	var reads [][]tio.Packet
+	var want []decodedSeg
+	seq := uint32(0)
+	for _, l := range large {
+		per, segs := 1, 3
+		if l {
+			per, segs = 2, 9
+		}
+		var b []tio.Packet
+		for range per {
+			b = append(b, flowTCPSuperpacket(0, seq, segs))
+			for s := range segs {
+				want = append(want, decodedSeg{seq: seq, idx: uint16(s)})
+			}
+			seq++
+		}
+		reads = append(reads, b)
+	}
+	return reads, want
+}
+
+func repeatBool(v bool, n int) []bool {
+	out := make([]bool, n)
+	for i := range out {
+		out[i] = v
+	}
+	return out
+}
+
+// TestFlowStagedTXGateSwitch forces the gate from inline to fanout and back, with a slow socket and slow
+// workers so the fanout batches are still in flight when the small reads arrive: wire order is read order, counters
+// are contiguous, and each switch happens on exactly the gate_hold-th read that wants it.
+func TestFlowStagedTXGateSwitch(t *testing.T) {
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+	const hold = 3
+	for _, routines := range []int{1, 2} {
+		t.Run(fmt.Sprintf("routines=%d", routines), func(t *testing.T) {
+			sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, gated(hold))
+			sender.f.routines = routines
+			sender.f.writers = []udp.Conn{sender.conn, sender.conn}
+			delayCrypto(sender, func(n uint64) {
+				if n%7 == 0 {
+					time.Sleep(time.Millisecond)
+				}
+			})
+			sender.conn.delay = 4 * time.Millisecond
+			// 4 small inline, 2 large held inline, 6 large fanout from the switch on, 2 small held in fanout, then
+			// the switch back drains and 3 small go inline
+			shape := append(repeatBool(false, 4), repeatBool(true, 8)...)
+			shape = append(shape, repeatBool(false, 5)...)
+			reads, want := gateReads(shape...)
+			ci := sender.hostinfo.ConnectionState
+			start := ci.messageCounter.Load()
+			c0 := readGateCounters()
+			p := newStagedTx(sender.f, newFlowBatchQueue(nil), 0)
+			for i, r := range reads {
+				p.send(r)
+				if i == 13 {
+					require.Positive(t, p.inFlight, "fanout batches must still be in flight when the small reads start")
+				}
+				if i == 14 {
+					require.Zero(t, p.inFlight, "switching to inline drains the pipeline")
+				}
+			}
+			p.close()
+			require.Equal(t, want, decryptWire(t, receiver, sender.conn.captured()), "wire order must match read order")
+			require.Equal(t, start+uint64(len(want)), ci.messageCounter.Load(), "every reserved counter is used once")
+			require.Equal(t, gateCounters{inline: 4 + 2 + 3, batches: 6 + 2, switches: 2}, readGateCounters().since(c0))
+			require.Zero(t, ci.txBudget.state.Load())
+		})
+	}
+}
+
+// TestFlowStagedTXGateHold: the gate never switches before gate_hold consecutive reads want the other mode, and
+// a read that wants the current mode restarts the count.
+func TestFlowStagedTXGateHold(t *testing.T) {
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+	const hold = 5
+	sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, gated(hold))
+	shape := append(repeatBool(true, hold-1), false)
+	shape = append(shape, repeatBool(true, hold-1)...)
+	reads, want := gateReads(append(shape, true)...)
+	c0 := readGateCounters()
+	p := newStagedTx(sender.f, newFlowBatchQueue(nil), 0)
+	for _, r := range reads[:len(reads)-1] {
+		p.send(r)
+	}
+	require.Equal(t, gateCounters{inline: 2*hold - 1}, readGateCounters().since(c0), "no switch before gate_hold")
+	p.send(reads[len(reads)-1])
+	require.Equal(t, gateCounters{inline: 2*hold - 1, batches: 1, switches: 1}, readGateCounters().since(c0))
+	p.close()
+	require.Equal(t, want, decryptWire(t, receiver, sender.conn.captured()))
+}

flow_test.go

@@ -358,9 +358,10 @@ func stagedOn(n int) stagedConfig {
 	return stagedConfig{workers: n, tx: true, rx: true, txSlots: stagedTxSlots, rxSlots: 6}
 }
 
-// adaptive is staging with the inline threshold parseStagedConfig sets.
+// adaptive is staging with the inline threshold and gate hold parseStagedConfig sets.
 func adaptive(n int) stagedConfig {
-	return stagedConfig{workers: n, tx: true, rx: true, inlineBelow: 256, txSlots: stagedTxSlots, rxSlots: 6}
+	return stagedConfig{workers: n, tx: true, rx: true, inlineBelow: 256, gateHold: stagedGateHold,
+		txSlots: stagedTxSlots, rxSlots: 6}
 }
 
 func stagedName(sc stagedConfig) string {

flow_tx.go

@@ -274,6 +274,11 @@ type stagedTx struct {
 	returned   chan *txSlot
 	inFlight   int
 	senderDone chan struct{}
+	// gateInline is the routine's transmit mode, gateStreak how many reads in a row wanted the other one
+	gateInline bool
+	gateStreak int
+
+	metricGateSwitches metrics.Counter
 
 	readBusy, encryptBusy, sendBusy metrics.Counter
 	metricBatches                   metrics.Counter
@@ -298,6 +303,9 @@ func newStagedTx(f *Interface, queue handoffQueue, q int) *stagedTx {
 		sendBusy:      stageBusy("tx_send", q),
 		metricBatches: metrics.GetOrRegisterCounter("flow.tx.batches", nil),
 		metricInline:  metrics.GetOrRegisterCounter("flow.tx.inline", nil),
+		// Starts inline, the pipeline is empty, unless the threshold never allows it
+		gateInline:         f.staged.inlineBelow > 0,
+		metricGateSwitches: metrics.GetOrRegisterCounter("flow.tx.gate_switches", nil),
 	}
 	if f.pinThreads {
 		p.cpu = f.threadCPU(q)
@@ -339,7 +347,7 @@ func (p *stagedTx) run(fwPacket *firewall.ParsedPacket, rejectBuf []byte, conntr
 			return err
 		}
 		// Waiting for a slot is backpressure, not reader work, so it comes before the clock read. The loop reads the
-		// clock twice, more only when it waits for tunnel credit or drains for an oversized read.
+		// clock twice, more only when it waits for tunnel credit or drains, for an oversized read or a switch to inline.
 		p.ready()
 		t := time.Now()
 		stall := p.handle(pkts, fwPacket, rejectBuf, conntrackCache.Get())
@@ -351,17 +359,16 @@ func (p *stagedTx) run(fwPacket *firewall.ParsedPacket, rejectBuf []byte, conntr
 // it itself, with how long it waited for tunnel credit.
 func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, rejectBuf []byte, localCache firewall.ConntrackCache) time.Duration {
 	p.ready()
-	inline := p.inFlight == 0 && p.small(pkts)
+	inline, stall := p.gate(p.small(pkts))
 	if inline && p.f.routines == 1 {
 		// With only one reader and an empty pipeline, no other staged reservation can be overtaken. Keep this common
 		// case inexpensive.
 		p.inlineSoleReader(pkts, fwPacket, rejectBuf, localCache)
-		return 0
+		return stall
 	}
 
 	slot := p.free[len(p.free)-1]
 	p.free = p.free[:len(p.free)-1]
-	var stall time.Duration
 	items := p.collect(slot.items[:0], pkts, fwPacket, rejectBuf, p.q, localCache)
 	if len(items) == 0 {
 		slot.items = items
@@ -411,6 +418,31 @@ func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, re
 	return stall
 }
 
+// gate reports whether a read goes inline, and how long it waited for the pipeline to drain. Each routine holds one
+// transmit mode, inline on the reader or fanout to the workers and the sender, until gateHold reads in a row want the
+// other one, the last of which goes the new way. Choosing per read let small inline sends leave from the reader's CPU
+// between staged ones leaving from the sender's, and the receiver's TCP saw them reordered. Going inline the reader
+// first waits until everything in flight is on the wire and the sender is idle, so no inline send overtakes a staged
+// one. Going to fanout nothing is in flight: inline sends finish before the reader moves on.
+func (p *stagedTx) gate(small bool) (bool, time.Duration) {
+	if small == p.gateInline {
+		p.gateStreak = 0
+		return small, 0
+	}
+	if p.gateStreak++; p.gateStreak < p.f.staged.gateHold {
+		return p.gateInline, 0
+	}
+	p.gateInline, p.gateStreak = small, 0
+	p.metricGateSwitches.Inc(1)
+	if !small {
+		return false, 0
+	}
+	t := time.Now()
+	for p.flushOldest() {
+	}
+	return true, time.Since(t)
+}
+
 // ready takes back every slot the sender has finished with, and waits for one when all of them are in flight.
 func (p *stagedTx) ready() {
 	p.reap()

flow_tx_test.go

@@ -295,6 +295,11 @@ func TestFlowStagedTXAffinity(t *testing.T) {
 						if pin && len(allowed) < 2 {
 							t.Skip("requires at least two allowed CPUs")
 						}
+						if sc.inlineBelow > 0 {
+							// Switch on every read that wants the other mode, so the small read goes inline between
+							// staged ones
+							sc.gateHold = 1
+						}
 						sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, sc)
 						f := sender.f
 						f.pinThreads, f.routines = pin, routines
@@ -417,20 +422,23 @@ func TestFlowTXBudgetManyReaders(t *testing.T) {
 	}
 }
 
-// TestFlowStagedTXInlineOnlyWhenEmpty: a small read behind a staged batch still in flight must not go inline, it
-// would overtake that batch.
+// TestFlowStagedTXInlineOnlyWhenEmpty: a small read behind a staged batch still in flight must not go inline before
+// that batch is on the wire, it would overtake it.
 func TestFlowStagedTXInlineOnlyWhenEmpty(t *testing.T) {
 	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
 	sc := adaptive(2)
-	sc.inlineBelow = 4
+	sc.inlineBelow, sc.gateHold = 4, 1
 	sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, sc)
 	p := newStagedTx(sender.f, newFlowBatchQueue(nil), 0)
 	block := make(chan struct{})
 	delayCrypto(sender, func(uint64) { <-block })
 	batches, want := flowTestBatches(1, 4, 1, 3)
-	p.send(batches[0][:3]) // its worker is blocked
-	p.send(batches[0][3:])
-	close(block)
+	p.send(batches[0][:3]) // switches to fanout, its worker is blocked
+	require.Equal(t, 1, p.inFlight)
+	unblock := time.AfterFunc(20*time.Millisecond, func() { close(block) })
+	defer unblock.Stop()
+	p.send(batches[0][3:]) // switches to inline, which waits for the staged batch first
+	require.Zero(t, p.inFlight)
 	p.close()
 	require.Equal(t, want, decryptWire(t, receiver, sender.conn.captured()))
 }
@@ -512,7 +520,10 @@ func TestFlowStagedTXAdaptiveBudget(t *testing.T) {
 	} {
 		t.Run(shape.name, func(t *testing.T) {
 			defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
-			sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, adaptive(2))
+			sc := adaptive(2)
+			// p0's large read must go to the workers at once, p1 stays inline
+			sc.gateHold = 1
+			sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, sc)
 			sender.f.routines = 2
 			sender.f.writers = []udp.Conn{sender.conn, sender.conn}
 			ci := sender.hostinfo.ConnectionState

main_test.go

@@ -142,7 +142,8 @@ func TestParseStagedConfig(t *testing.T) {
 		require.Equal(t, stagedConfig{}, staged(2))
 		return
 	}
-	want := stagedConfig{workers: 2, tx: !noiseutil.EncryptLockNeeded, rx: true, inlineBelow: 256, txSlots: 8, rxSlots: 6}
+	want := stagedConfig{workers: 2, tx: !noiseutil.EncryptLockNeeded, rx: true, inlineBelow: 256, gateHold: stagedGateHold,
+		txSlots: 8, rxSlots: 6}
 	require.Equal(t, want, staged(2))
 
 	c := config.NewC(l)
@@ -163,6 +164,19 @@ func TestParseStagedConfig(t *testing.T) {
 	require.Equal(t, 32, parseStagedConfig(c, l).inlineBelow)
 	c.Settings["tun"] = map[string]any{"staged": map[string]any{"workers": 2, "inline_below": -1}}
 	require.Zero(t, parseStagedConfig(c, l).inlineBelow, "a negative threshold is 0")
+
+	for _, tt := range []struct {
+		hold any
+		want int
+	}{{nil, stagedGateHold}, {8, 8}, {0, 1}, {-3, 1}} {
+		c = config.NewC(l)
+		settings := map[string]any{"workers": 2}
+		if tt.hold != nil {
+			settings["gate_hold"] = tt.hold
+		}
+		c.Settings["tun"] = map[string]any{"staged": settings}
+		require.Equal(t, tt.want, parseStagedConfig(c, l).gateHold, "gate_hold %v", tt.hold)
+	}
 }
 
 func TestStagedSlots(t *testing.T) {