diff --git a/e2e/tunnels_test.go b/e2e/tunnels_test.go index 4117dd83..b7c3ed0b 100644 --- a/e2e/tunnels_test.go +++ b/e2e/tunnels_test.go @@ -578,7 +578,7 @@ func TestCloseTunnelAuthenticated(t *testing.T) { // Other platforms must do the same over the inline fallback. func TestStagedTunnel(t *testing.T) { t.Parallel() - received := metrics.GetOrRegisterCounter("flow.rx.batches", nil) + received := metrics.GetOrRegisterCounter("flow.rx.inline", nil) before := received.Count() staged := m{"tun": m{"staged": m{"workers": 2}}} ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{}) diff --git a/examples/config.yml b/examples/config.yml index 52c8cb57..51630b2e 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -284,6 +284,8 @@ 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. #staged: # workers: 0 @@ -291,6 +293,7 @@ tun: # batched sends flowing through one XPS-selected NIC TX ring, so packets within a flow stay ordered on the wire # instead of being sprayed across multiple TX rings and reordered. Not reloadable. Coerced to false if routines <= 1. # Staged senders are only pinned with an explicit cpu_affinity, see below; otherwise the scheduler places them. + # Small batches a reader sends itself can use a different TX ring than its sender. #pin_threads: true # pin_threads_key helps the CPU-auto-selector shuffle which CPUs are chosen for pinning. diff --git a/flow.go b/flow.go index dfca2109..d6e9772a 100644 --- a/flow.go +++ b/flow.go @@ -20,16 +20,19 @@ import ( // with GRO their buffers take about 9 MiB and 24 MiB per routine. 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.inline oversized reads the reader sent itself. 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. Small RX batches sent to the writer include decryption in writer time. +// 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. +// 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. 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 int } // parseStagedConfig enables staging on Linux only; queues and conns that cannot hand off batches also stay inline. @@ -43,6 +46,8 @@ func parseStagedConfig(c *config.C, l *slog.Logger) stagedConfig { return stagedConfig{} } sc.tx, sc.rx = true, true + // 256 is the measured crossover + sc.inlineBelow = 256 if noiseutil.EncryptLockNeeded { l.Warn("tun.staged cannot encrypt in parallel with boringcrypto or fips140; transmit stays inline") sc.tx = false diff --git a/flow_rx.go b/flow_rx.go index b4500bce..2c98b8f1 100644 --- a/flow_rx.go +++ b/flow_rx.go @@ -98,6 +98,7 @@ type stagedRx struct { readBusy, decryptBusy, writeBusy metrics.Counter metricBatches metrics.Counter + metricInline metrics.Counter } func newStagedRx(f *Interface, q int, wrxc *rxContext) *stagedRx { @@ -114,6 +115,7 @@ func newStagedRx(f *Interface, q int, wrxc *rxContext) *stagedRx { decryptBusy: stageBusy("rx_decrypt", q), writeBusy: stageBusy("rx_write", q), metricBatches: metrics.GetOrRegisterCounter("flow.rx.batches", nil), + metricInline: metrics.GetOrRegisterCounter("flow.rx.inline", nil), } workers := f.staged.workers p.pool = newFlowPool(workers, stagedRxSlots*2*workers, p.decryptBusy) @@ -234,6 +236,14 @@ func (p *stagedRx) endBatch() { return } + if p.inFlight == 0 && n < p.f.staged.inlineBelow { + // The writer is idle with nothing in flight, so the listener may commit with its context and flush the tun + p.metricInline.Inc(1) + p.process(b, p.wrxc) + b.reset() + return + } + parts := min(len(b.chunks), b.data/rxChunkMin) p.metricBatches.Inc(1) if parts > 0 { diff --git a/flow_rx_test.go b/flow_rx_test.go index a131ab03..36a0f627 100644 --- a/flow_rx_test.go +++ b/flow_rx_test.go @@ -10,6 +10,7 @@ import ( "time" "github.com/flynn/noise" + "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/header" "github.com/slackhq/nebula/noiseutil" "github.com/slackhq/nebula/udp" @@ -92,7 +93,7 @@ func sendControl(sender *flowTestPeer, typ header.MessageType, st header.Message // receivePaths is inline, staged and, with the threshold, adaptive receive. func receivePaths() []stagedConfig { - return []stagedConfig{{}, stagedOn(2)} + return []stagedConfig{{}, stagedOn(2), adaptive(2)} } // TestFlowStagedRXRecvError feeds a RecvError that deletes the tunnel, then a known replay and new data for it, in one @@ -218,3 +219,19 @@ func TestFlowStagedRXCloseFlushesPartial(t *testing.T) { p.close() require.Equal(t, want, receiver.tun.segments(t)) } + +// TestFlowStagedRXInlineBelow: small idle batches stay on the listener, larger batches still use the ordered writer. +func TestFlowStagedRXInlineBelow(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + sc := adaptive(3) + sc.inlineBelow = 16 + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, sc) + batches, want := flowTestBatches(40, 4, 1, 16) + wire := wireFor(t, sender, batches) + inline := metrics.GetOrRegisterCounter("flow.rx.inline", nil) + inline0, rx0 := inline.Count(), rxBatches.Count() + receiver.receive(&rxConn{wire: wire, sizes: []int{8, 8, 8, 128, 128, 8, 8, 128, 4, 128}}) + require.Equal(t, want, receiver.tun.segments(t)) + require.Positive(t, inline.Count()-inline0) + require.Positive(t, rxBatches.Count()-rx0) +} diff --git a/flow_test.go b/flow_test.go index 8e059d59..ecd24429 100644 --- a/flow_test.go +++ b/flow_test.go @@ -358,10 +358,17 @@ func stagedOn(n int) stagedConfig { return stagedConfig{workers: n, tx: true, rx: true} } +// adaptive is staging with the inline threshold parseStagedConfig sets. +func adaptive(n int) stagedConfig { + return stagedConfig{workers: n, tx: true, rx: true, inlineBelow: 256} +} + func stagedName(sc stagedConfig) string { switch { case sc.workers == 0: return "inline" + case sc.inlineBelow > 0: + return "adaptive" default: return "staged" } diff --git a/flow_tx.go b/flow_tx.go index 80ccbb54..3313ffe5 100644 --- a/flow_tx.go +++ b/flow_tx.go @@ -270,7 +270,7 @@ func (s *txSlot) run(nb []byte) { type stagedTx struct { txReserver q int - cpu int // reader CPU + cpu int // reader CPU, inline sends stay here senderCPU int // -1 leaves the sender within its inherited affinity queue handoffQueue pool *flowPool @@ -359,6 +359,13 @@ func (p *stagedTx) run(fwPacket *firewall.ParsedPacket, rejectBuf []byte, conntr // it itself, with how long it waited for the sender. func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, rejectBuf []byte, localCache firewall.ConntrackCache) time.Duration { p.reap() + inline := p.inFlight == 0 && 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 + } slot, stall := p.acquire() items := p.collect(slot.items[:0], pkts, fwPacket, rejectBuf, p.q, localCache) @@ -392,6 +399,12 @@ func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, re return stall } + if inline { + p.inline(slot) + p.free = append(p.free, slot) + return stall + } + // The packets now live in the slot, and the queue reads into the slot's old buffer slot.buf = p.queue.SwapRxBuf(slot.buf) @@ -474,3 +487,47 @@ func (p *stagedTx) close() { <-p.senderDone p.pool.close() } + +// small counts segments against the inline threshold. +func (p *stagedTx) small(pkts []tio.Packet) bool { + below := uint64(p.f.staged.inlineBelow) + if below == 0 { + return false + } + var n uint64 + for _, pkt := range pkts { + if n += segmentCount(pkt); n >= below { + return false + } + } + return true +} + +// inline encrypts a budgeted batch on the reader, with no worker handoff or buffer swap. The local sender is idle, +// but other readers may hold reservations on these tunnels, so keep our credit until the flush completes. +func (p *stagedTx) inline(slot *txSlot) { + p.metricInline.Inc(1) + for i := range slot.items { + p.f.encryptTxItem(&slot.items[i], p.nb, p.sb) + if p.sb.Len() >= batch.SendBatchCap { + p.f.flushSendBatch(p.sb, p.q) + } + } + p.f.flushSendBatch(p.sb, p.q) + slot.releaseBudget() + clear(slot.items) + slot.items = slot.items[:0] +} + +// inlineSoleReader needs no credit: the interface has no other tun reader and this reader has drained every +// reservation before taking the inline path. +func (p *stagedTx) inlineSoleReader(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, rejectBuf []byte, localCache firewall.ConntrackCache) { + p.metricInline.Inc(1) + for _, pkt := range pkts { + p.f.consumeInsidePacket(pkt, fwPacket, p.nb, p.sb, rejectBuf, p.q, localCache) + if p.sb.Len() >= batch.SendBatchCap { + p.f.flushSendBatch(p.sb, p.q) + } + } + p.f.flushSendBatch(p.sb, p.q) +} diff --git a/flow_tx_test.go b/flow_tx_test.go index 7dabad48..c6877a73 100644 --- a/flow_tx_test.go +++ b/flow_tx_test.go @@ -272,6 +272,7 @@ func requireFlowCPU(t *testing.T, masks []unix.CPUSet, cpu int, allowed []int) { // TestFlowStagedTXAffinity sends large, small and large reads, each from an idle pipeline, and checks the order on the // wire and where the reader and each send actually ran, for every CPU placement, with one routine and with several. A // staged sender is only pinned to an explicit entry other than its reader's CPU, and inline sends stay on the reader. +// With the threshold, only the small read goes inline. // The reader only reads again once its earlier reads are on the wire, so this also checks the sender never waits for // the reader. Pinning threads changes the process, so the test runs in a child. func TestFlowStagedTXAffinity(t *testing.T) { @@ -316,14 +317,22 @@ func TestFlowStagedTXAffinity(t *testing.T) { } } smallCPU := sendCPU + if sc.inlineBelow > 0 { + smallCPU = readerCPU + } conn := &affinityConn{} f.writers = []udp.Conn{conn} batches, want := flowTestBatches(3, 8, 1, 45) batches[1] = batches[1][:1] want = append(want[:405], want[720:]...) q := &affinityQueue{flowBatchQueue: newFlowBatchQueue(batches), conn: conn} + inline := metrics.GetOrRegisterCounter("flow.tx.inline", nil) + inline0 := inline.Count() f.listenIn(q, 0) require.Equal(t, want, decryptWire(t, receiver, conn.captured())) + if sc.inlineBelow > 0 { + require.Equal(t, int64(1), inline.Count()-inline0) + } b := q.boundaries requireFlowCPU(t, q.masks, readerCPU, allowed) requireFlowCPU(t, conn.masks[:b[1]], sendCPU, allowed) @@ -338,7 +347,7 @@ func TestFlowStagedTXAffinity(t *testing.T) { // sendPaths is inline, staged and, with the threshold, adaptive transmit. func sendPaths() []stagedConfig { - return []stagedConfig{{}, stagedOn(2)} + return []stagedConfig{{}, stagedOn(2), adaptive(2)} } // requireAccepted checks the receiver accepts every packet of wire in the order given. With several readers the wire @@ -405,3 +414,85 @@ 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. +func TestFlowStagedTXInlineOnlyWhenEmpty(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + sc := adaptive(2) + sc.inlineBelow = 4 + 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.close() + require.Equal(t, want, decryptWire(t, receiver, sender.conn.captured())) +} + +// 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) { + for _, shape := range []struct { + name string + packets, segments int + }{ + {"single", 1, 1}, {"batch", 8, 30}, + } { + t.Run(shape.name, func(t *testing.T) { + defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) + sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, adaptive(2)) + sender.f.routines = 2 + sender.f.writers = []udp.Conn{sender.conn, sender.conn} + ci := sender.hostinfo.ConnectionState + start := ci.messageCounter.Load() + entered, unblock := make(chan struct{}), make(chan struct{}) + delayCrypto(sender, func(n uint64) { + if n == start+1 { + close(entered) + <-unblock + } + }) + p0 := newStagedTx(sender.f, newFlowBatchQueue(nil), 0) + p1 := newStagedTx(sender.f, newFlowBatchQueue(nil), 1) + defer p0.close() + defer p1.close() + var release sync.Once + defer release.Do(func() { close(unblock) }) + large, _ := flowTestBatches(1, 8, 1, 45) + p0.send(large[0]) + <-entered + perBatch := shape.packets * shape.segments + small, _ := flowTestBatches(9600/perBatch, shape.packets, 1, shape.segments) + fits := int(ci.txBudgetLimit()-360) / perBatch + for _, packets := range small[:fits] { + p1.send(packets) + } + require.Equal(t, uint64(360+fits*perBatch), (ci.txBudget.state.Load()>>24)&txBudgetMask, "inline sends must spend shared credit") + done := make(chan struct{}) + go func() { + defer close(done) + for _, packets := range small[fits:] { + p1.send(packets) + } + }() + select { + case <-done: + t.Error("adaptive reader bypassed outstanding credit") + case <-time.After(50 * time.Millisecond): + } + require.LessOrEqual(t, ci.messageCounter.Load()-start, ci.txBudgetLimit()) + release.Do(func() { close(unblock) }) + <-done + p0.drain() + p1.drain() + wire := sender.conn.captured() + require.Len(t, wire, 360+40*240) + requireAccepted(t, receiver, wire) + require.Zero(t, ci.txBudget.state.Load()) + }) + } +} diff --git a/main_test.go b/main_test.go index 2149a105..04be2459 100644 --- a/main_test.go +++ b/main_test.go @@ -142,6 +142,6 @@ func TestParseStagedConfig(t *testing.T) { require.Equal(t, stagedConfig{}, staged(2)) return } - want := stagedConfig{workers: 2, tx: !noiseutil.EncryptLockNeeded, rx: true} + want := stagedConfig{workers: 2, tx: !noiseutil.EncryptLockNeeded, rx: true, inlineBelow: 256} require.Equal(t, want, staged(2)) }