From d16a52f9212ac2e353011d88fc1ffe7cfc95ed17 Mon Sep 17 00:00:00 2001 From: Ryan Huber Date: Thu, 15 Oct 2026 20:53:47 +0000 Subject: [PATCH] tun: Add tun.staged.tx and tun.staged.inline_below tx: false keeps transmit inline and stages only receive. inline_below sets the idle-reader shortcut threshold, 0 hands every batch to the workers and the ordered sender or writer. --- examples/config.yml | 8 ++++-- flow.go | 12 ++++---- flow_tx_test.go | 68 +++++++++++++++++++++++++++++++++++++++++++++ main_test.go | 13 +++++++++ 4 files changed, 93 insertions(+), 8 deletions(-) diff --git a/examples/config.yml b/examples/config.yml index 4c0717bc..561a0789 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -284,14 +284,18 @@ 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 256 TX segments or RX packets stay on the reader. With multiple routines, those sends still - # take tunnel reservation credit. + # 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. + # 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 # recvmmsg batch, listen.batch * 64 KiB with udp_offloads or * 9 KiB without; the default keeps up to 384 receive # entries in flight in 4 to 6 slots: 6 slots or 24 MiB at the default listen.batch of 64. Lower them to save memory. #staged: # workers: 0 + # tx: true + # inline_below: 256 # tx_slots: 8 # rx_slots: 6 diff --git a/flow.go b/flow.go index 92e22407..328f0e38 100644 --- a/flow.go +++ b/flow.go @@ -34,7 +34,7 @@ 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 + // inlineBelow is the TX segment or RX packet count below which an idle reader skips the worker handoff, 0 never inlineBelow int // txSlots and rxSlots are how many batches each routine can have in flight per direction txSlots, rxSlots int @@ -78,17 +78,17 @@ func parseStagedConfig(c *config.C, l *slog.Logger) stagedConfig { l.Warn("tun.staged is only supported on Linux; packet processing stays inline") return stagedConfig{} } - sc.tx, sc.rx = true, true - // 256 is the measured crossover - sc.inlineBelow = 256 - if noiseutil.EncryptLockNeeded { + sc.tx, sc.rx = c.GetBool("tun.staged.tx", true), true + // 256 is the measured crossover, 0 hands every batch to the workers + sc.inlineBelow = max(c.GetInt("tun.staged.inline_below", 256), 0) + if sc.tx && noiseutil.EncryptLockNeeded { l.Warn("tun.staged cannot encrypt in parallel with boringcrypto or fips140; transmit stays inline") sc.tx = false } 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, - "txSlots", sc.txSlots, "rxSlots", sc.rxSlots) + "inlineBelow", sc.inlineBelow, "txSlots", sc.txSlots, "rxSlots", sc.rxSlots) return sc } diff --git a/flow_tx_test.go b/flow_tx_test.go index 8a96eab6..3262492b 100644 --- a/flow_tx_test.go +++ b/flow_tx_test.go @@ -14,12 +14,14 @@ import ( "github.com/flynn/noise" "github.com/rcrowley/go-metrics" + "github.com/slackhq/nebula/config" "github.com/slackhq/nebula/cpupick" "github.com/slackhq/nebula/firewall" "github.com/slackhq/nebula/header" "github.com/slackhq/nebula/noiseutil" "github.com/slackhq/nebula/overlay/batch" "github.com/slackhq/nebula/overlay/tio" + "github.com/slackhq/nebula/test" "github.com/slackhq/nebula/udp" "github.com/slackhq/nebula/util" "github.com/stretchr/testify/require" @@ -433,6 +435,72 @@ func TestFlowStagedTXInlineOnlyWhenEmpty(t *testing.T) { require.Equal(t, want, decryptWire(t, receiver, sender.conn.captured())) } +// TestFlowStagedTXInlineBelowZero: with inline_below 0 an idle reader never sends a batch itself, in either +// direction: every read goes through the workers and the ordered sender or writer. +func TestFlowStagedTXInlineBelowZero(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + l := test.NewLogger() + c := config.NewC(l) + c.Settings = map[string]any{"tun": map[string]any{"staged": map[string]any{"workers": 2, "inline_below": 0}}} + sc := parseStagedConfig(c, l) + if !sc.tx { + t.Skip("transmit stays inline in this build") + } + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, sc) + batches, want := flowTestBatches(20, 2, 1, 4) + txInline := metrics.GetOrRegisterCounter("flow.tx.inline", nil) + tx := metrics.GetOrRegisterCounter("flow.tx.batches", nil) + rxInline := metrics.GetOrRegisterCounter("flow.rx.inline", nil) + txInline0, tx0, rxInline0, rx0 := txInline.Count(), tx.Count(), rxInline.Count(), rxBatches.Count() + sender.f.listenIn(newFlowBatchQueue(batches), 0) + wire := sender.conn.captured() + require.Equal(t, want, decryptWire(t, receiver, wire)) + require.Zero(t, txInline.Count()-txInline0) + require.Equal(t, int64(len(batches)), tx.Count()-tx0) + receiver.receive(&rxConn{wire: wire, sizes: []int{1, 2, 4}}) + require.Equal(t, want, receiver.tun.segments(t)) + require.Zero(t, rxInline.Count()-rxInline0) + require.Positive(t, rxBatches.Count()-rx0) +} + +// TestFlowStagedTXDisabled: with tx false the reader never builds a staged pipeline, even on a queue that could hand +// off its batches. Receive stays staged. +func TestFlowStagedTXDisabled(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + l := test.NewLogger() + c := config.NewC(l) + c.Settings = map[string]any{"tun": map[string]any{"staged": map[string]any{"workers": 2, "tx": false}}} + sc := parseStagedConfig(c, l) + require.False(t, sc.tx) + require.True(t, sc.rx) + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, sc) + batches, want := flowTestBatches(4, 8, 1, 45) + tx := metrics.GetOrRegisterCounter("flow.tx.batches", nil) + txInline := metrics.GetOrRegisterCounter("flow.tx.inline", nil) + tx0, txInline0, rx0 := tx.Count(), txInline.Count(), rxBatches.Count() + q := &swapCountQueue{flowBatchQueue: newFlowBatchQueue(batches)} + sender.f.listenIn(q, 0) + wire := sender.conn.captured() + require.Equal(t, want, decryptWire(t, receiver, wire)) + require.Zero(t, q.swaps, "newStagedTx always takes the queue's buffer, so it never ran") + require.Equal(t, tx0, tx.Count()) + require.Equal(t, txInline0, txInline.Count()) + receiver.receive(&rxConn{wire: wire, batchLen: 512}) + require.Equal(t, want, receiver.tun.segments(t)) + require.Greater(t, rxBatches.Count(), rx0) +} + +// swapCountQueue counts SwapRxBuf calls. +type swapCountQueue struct { + *flowBatchQueue + swaps int +} + +func (q *swapCountQueue) SwapRxBuf(buf []byte) []byte { + q.swaps++ + return q.flowBatchQueue.SwapRxBuf(buf) +} + // TestFlowStagedTXAdaptiveBudget: a fast reader's adaptive sends must stop spending credit until another reader's // earlier reservation reaches the wire. func TestFlowStagedTXAdaptiveBudget(t *testing.T) { diff --git a/main_test.go b/main_test.go index db6641b2..1edf260a 100644 --- a/main_test.go +++ b/main_test.go @@ -150,6 +150,19 @@ func TestParseStagedConfig(t *testing.T) { c.Settings["tun"] = map[string]any{"staged": map[string]any{"workers": 2, "tx_slots": 3, "rx_slots": 1}} got := parseStagedConfig(c, l) require.Equal(t, []int{3, 2}, []int{got.txSlots, got.rxSlots}, "overrides apply, down to the minimum") + + c = config.NewC(l) + c.Settings["tun"] = map[string]any{"staged": map[string]any{"workers": 2, "tx": false, "inline_below": 0}} + got = parseStagedConfig(c, l) + require.False(t, got.tx, "tx: false keeps transmit inline") + require.True(t, got.rx) + require.Zero(t, got.inlineBelow, "inline_below: 0 hands every batch off") + + c = config.NewC(l) + c.Settings["tun"] = map[string]any{"staged": map[string]any{"workers": 2, "inline_below": 32}} + 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") } func TestStagedSlots(t *testing.T) {