diff --git a/examples/config.yml b/examples/config.yml index 561a0789..9be1436f 100644 --- a/examples/config.yml +++ b/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 diff --git a/flow.go b/flow.go index 328f0e38..3b9d00ae 100644 --- a/flow.go +++ b/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..q.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..q.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 } diff --git a/flow_gate_test.go b/flow_gate_test.go new file mode 100644 index 00000000..f713d06c --- /dev/null +++ b/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())) +} diff --git a/flow_test.go b/flow_test.go index 4eb67a2e..9e82bfed 100644 --- a/flow_test.go +++ b/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 { diff --git a/flow_tx.go b/flow_tx.go index 0df643bf..0b7b5175 100644 --- a/flow_tx.go +++ b/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() diff --git a/flow_tx_test.go b/flow_tx_test.go index 3262492b..cf58db76 100644 --- a/flow_tx_test.go +++ b/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 diff --git a/main_test.go b/main_test.go index 1edf260a..1980de9c 100644 --- a/main_test.go +++ b/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) {