From 5cbbba254ea5d5e7d1e940b91107824d0c4464f3 Mon Sep 17 00:00:00 2001 From: Ryan Huber Date: Fri, 16 Oct 2026 21:20:43 +0000 Subject: [PATCH] tun: Drop the staged timing metrics --- flow.go | 32 ++-------------------------- flow_budget.go | 18 ++++++---------- flow_rx.go | 47 +++++++---------------------------------- flow_test.go | 31 --------------------------- flow_tx.go | 57 +++++++++++++++++++------------------------------- 5 files changed, 37 insertions(+), 148 deletions(-) diff --git a/flow.go b/flow.go index 3b9d00ae..c23d98a1 100644 --- a/flow.go +++ b/flow.go @@ -1,13 +1,10 @@ package nebula import ( - "fmt" "log/slog" "runtime" "sync" - "time" - "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/config" "github.com/slackhq/nebula/noiseutil" ) @@ -25,10 +22,6 @@ import ( // Metrics are registered only for active pipelines. flow.{tx,rx}.batches counts reads handed to the ordered stage and // 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. type stagedConfig struct { @@ -112,38 +105,17 @@ type flowPool struct { wg sync.WaitGroup } -func newFlowPool(workers, queueLen int, busy metrics.Counter) *flowPool { +func newFlowPool(workers, queueLen int) *flowPool { p := &flowPool{jobs: make(chan flowJob, queueLen)} for range workers { p.wg.Go(func() { nb := make([]byte, 12) for job := range p.jobs { - // One pair of clock reads covers a run of jobs that are queued back to back - t := time.Now() - for job != nil { - job.run(nb) - job = p.next() - } - busy.Inc(int64(time.Since(t))) + job.run(nb) } }) } return p } -// next returns the next queued job without waiting, or nil. -func (p *flowPool) next() flowJob { - select { - case job := <-p.jobs: - return job - default: - return nil - } -} - func (p *flowPool) close() { close(p.jobs); p.wg.Wait() } - -// stageBusy is the busy time counter of one stage of routine q. -func stageBusy(stage string, q int) metrics.Counter { - return metrics.GetOrRegisterCounter(fmt.Sprintf("flow.stage.%s.q%d.busy_ns", stage, q), nil) -} diff --git a/flow_budget.go b/flow_budget.go index 0f8d8279..e75289ba 100644 --- a/flow_budget.go +++ b/flow_budget.go @@ -112,20 +112,19 @@ func (r *txReserver) tryBudget() bool { // orders. The sleep is only for readers with no local work, not the uncontended hot path. Acquisition is // all-or-nothing per read, so pressure on one tunnel also delays packets for idle tunnels in the read; splitting // would need separate slot ownership and coupled inner/relay reservations. Reader fairness is not guaranteed. -func (p *stagedTx) waitBudget() (time.Duration, bool) { +func (p *stagedTx) waitBudget() bool { if p.tryBudget() { - return 0, true + return true } - t := time.Now() for { if p.f.closed.Load() { - return time.Since(t), false + return false } if !p.flushOldest() { time.Sleep(50 * time.Microsecond) } if p.tryBudget() { - return time.Since(t), true + return true } } } @@ -142,11 +141,9 @@ func (s *txSlot) releaseBudget() { // the pipeline, then segments and sends them synchronously with credit for each segment, never exceeding the cap and // never filtering or running connection manager side effects twice. This deliberately slow path avoids a lone-batch // exception to the shared invariant. -func (p *stagedTx) sendLarge(items []txItem) time.Duration { - t := time.Now() +func (p *stagedTx) sendLarge(items []txItem) { for p.flushOldest() { } - stall := time.Since(t) for _, it := range items { err := tio.SegmentSuperpacket(it.pkt, func(seg []byte) error { if p.f.closed.Load() { @@ -159,9 +156,7 @@ func (p *stagedTx) sendLarge(items []txItem) time.Duration { if part.relayHI != nil { part.relayCounter = p.need(part.relayHI, part.relayHI.ConnectionState, 1) } - d, ok := p.waitBudget() - stall += d - if !ok { + if !p.waitBudget() { return nil } one := [1]txItem{part} @@ -179,5 +174,4 @@ func (p *stagedTx) sendLarge(items []txItem) time.Duration { } } clear(items) - return stall } diff --git a/flow_rx.go b/flow_rx.go index e0b5100c..a564df58 100644 --- a/flow_rx.go +++ b/flow_rx.go @@ -3,7 +3,6 @@ package nebula import ( "encoding/binary" "net/netip" - "time" "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/header" @@ -87,7 +86,6 @@ type stagedRx struct { hostmapCache map[uint32]*HostInfo wrxc *rxContext - started time.Time cur *rxBatch free []*rxBatch ordered chan *rxBatch @@ -95,9 +93,8 @@ type stagedRx struct { inFlight int writerDone chan struct{} - readBusy, decryptBusy, writeBusy metrics.Counter - metricBatches metrics.Counter - metricInline metrics.Counter + metricBatches metrics.Counter + metricInline metrics.Counter } func newStagedRx(f *Interface, q int, wrxc *rxContext) *stagedRx { @@ -110,14 +107,11 @@ func newStagedRx(f *Interface, q int, wrxc *rxContext) *stagedRx { ordered: make(chan *rxBatch, f.staged.rxSlots), returned: make(chan *rxBatch, f.staged.rxSlots), writerDone: make(chan struct{}), - readBusy: stageBusy("rx_read", q), - 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, f.staged.rxSlots*2*workers, p.decryptBusy) + p.pool = newFlowPool(workers, f.staged.rxSlots*2*workers) for range f.staged.rxSlots { // Two chunks per worker, so the writer can start committing while the workers are still on the rest b := &rxBatch{items: make([]rxItem, 0, 256), chunks: make([]rxChunk, 2*workers)} @@ -142,9 +136,6 @@ func (p *stagedRx) listen(li handoffConn) error { // deliver is the listener's per-packet work: classify the packet and add it to the current batch. Nothing here // decrypts or touches the tun. func (p *stagedRx) deliver(via netip.AddrPort, packet []byte) { - if p.started.IsZero() { - p.started = time.Now() - } b := p.cur it := rxItem{packet: packet, via: via} p.classify(&it) @@ -209,7 +200,6 @@ func (p *stagedRx) precheck(b *rxBatch) { func (p *stagedRx) handoff(bufs [][]byte) [][]byte { p.cur.bufs = bufs p.endBatch() - p.stopClock() b := p.cur if len(b.bufs) != len(bufs) || len(b.bufs[0]) < len(bufs[0]) { b.bufs = make([][]byte, len(bufs)) @@ -258,24 +248,13 @@ func (p *stagedRx) endBatch() { p.ordered <- b if len(p.free) == 0 { - // Every batch is in flight, wait for the writer: the socket buffer absorbs the backlog meanwhile. The wait is - // downstream backpressure, so the listener's busy time ends here. - p.stopClock() + // Every batch is in flight, wait for the writer: the socket buffer absorbs the backlog meanwhile. p.release(<-p.returned) } p.cur = p.free[len(p.free)-1] p.free = p.free[:len(p.free)-1] } -// stopClock adds the listener's busy time since the first packet of the batch, once per batch: the clock is read -// when that packet arrives and when the batch is handed off or has to wait. -func (p *stagedRx) stopClock() { - if !p.started.IsZero() { - p.readBusy.Inc(int64(time.Since(p.started))) - p.started = time.Time{} - } -} - // reap takes back every batch the writer has finished with, without waiting. func (p *stagedRx) reap() { for { @@ -294,9 +273,8 @@ func (p *stagedRx) release(b *rxBatch) { } // process is the ordered stage for one batch: decrypt whatever no worker is decrypting, commit every item in arrival -// order, and flush the tun batcher. It reports how long it waited on workers. -func (p *stagedRx) process(b *rxBatch, rxc *rxContext) time.Duration { - var waited time.Duration +// order, and flush the tun batcher. +func (p *stagedRx) process(b *rxBatch, rxc *rxContext) { if b.nChunks == 0 { decryptRxItems(b.items, rxc.nb) p.commit(b.items, rxc) @@ -304,18 +282,11 @@ func (p *stagedRx) process(b *rxBatch, rxc *rxContext) time.Duration { // Commit chunk by chunk, so the commit overlaps the workers still on later chunks for i := range b.nChunks { c := &b.chunks[i] - select { - case <-c.done: - default: - t := time.Now() - <-c.done - waited += time.Since(t) - } + <-c.done p.commit(c.items, rxc) } } p.f.flushOutside(rxc) - return waited } // commit runs the ordered stage over items in arrival order. A non-data packet goes through readOutsidePackets in @@ -382,9 +353,7 @@ func (p *stagedRx) writer() { <-b.chunks[i].done } } else { - t := time.Now() - waited := p.process(b, p.wrxc) - p.writeBusy.Inc(int64(time.Since(t) - waited)) + p.process(b, p.wrxc) } b.reset() p.returned <- b diff --git a/flow_test.go b/flow_test.go index 9e82bfed..15ef4081 100644 --- a/flow_test.go +++ b/flow_test.go @@ -534,34 +534,3 @@ func TestFlowStagedCipherFallback(t *testing.T) { require.Greater(t, rxBatches.Count(), rx0) require.Equal(t, want, receiver.tun.segments(t)) } - -// TestFlowStagedBusyExcludesBackpressure runs both directions with a slow downstream stage and only two slots, so the -// reader keeps waiting for one: the reader's busy time must leave those waits out. -func TestFlowStagedBusyExcludesBackpressure(t *testing.T) { - defer goleak.VerifyNone(t, goleak.IgnoreCurrent()) - const delay = 20 * time.Millisecond - sc := stagedOn(2) - sc.txSlots, sc.rxSlots = 2, 2 - sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, sc) - batches, want := flowTestBatches(10, 2, 1, 8) - - sender.conn.delay = delay - txRead, txSend := stageBusy("tx_read", 0), stageBusy("tx_send", 0) - read0, send0 := txRead.Count(), txSend.Count() - start := time.Now() - sender.f.listenIn(newFlowBatchQueue(batches), 0) - elapsed := time.Since(start) - require.GreaterOrEqual(t, elapsed, time.Duration(len(batches)-2)*delay, "the reader must have waited for slots") - require.GreaterOrEqual(t, time.Duration(txSend.Count()-send0), time.Duration(len(batches))*delay) - require.Less(t, time.Duration(txRead.Count()-read0), elapsed/2, "tx_read must not count slot waits") - - receiver.tun.delay = delay / 10 - rxRead := stageBusy("rx_read", 0) - read0 = rxRead.Count() - start = time.Now() - receiver.receive(&rxConn{wire: sender.conn.captured(), batchLen: 16}) - elapsed = time.Since(start) - require.Equal(t, want, receiver.tun.segments(t)) - require.GreaterOrEqual(t, elapsed, 3*delay/10, "the listener must have waited for batches") - require.Less(t, time.Duration(rxRead.Count()-read0), elapsed/2, "rx_read must not count batch waits") -} diff --git a/flow_tx.go b/flow_tx.go index 0b7b5175..edf09592 100644 --- a/flow_tx.go +++ b/flow_tx.go @@ -3,7 +3,6 @@ package nebula import ( "errors" "net/netip" - "time" "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/firewall" @@ -280,9 +279,8 @@ type stagedTx struct { metricGateSwitches metrics.Counter - readBusy, encryptBusy, sendBusy metrics.Counter - metricBatches metrics.Counter - metricInline metrics.Counter + metricBatches metrics.Counter + metricInline metrics.Counter } // newStagedTx needs a queue that hands over whole read batches, the offload (TSO/USO) queue on linux. On any other @@ -298,9 +296,6 @@ func newStagedTx(f *Interface, queue handoffQueue, q int) *stagedTx { ordered: make(chan *txSlot, f.staged.txSlots), returned: make(chan *txSlot, f.staged.txSlots), senderDone: make(chan struct{}), - readBusy: stageBusy("tx_read", q), - encryptBusy: stageBusy("tx_encrypt", q), - 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 @@ -315,7 +310,7 @@ func newStagedTx(f *Interface, queue handoffQueue, q int) *stagedTx { p.senderCPU = f.cpuAffinity[(q+1)%n] } } - p.pool = newFlowPool(f.staged.workers, f.staged.txSlots, p.encryptBusy) + p.pool = newFlowPool(f.staged.workers, f.staged.txSlots) // Ciphertext adds per-segment headers and tags; the arena can grow for small segments. buf := queue.SwapRxBuf(nil) @@ -346,25 +341,21 @@ func (p *stagedTx) run(fwPacket *firewall.ParsedPacket, rejectBuf []byte, conntr if err != nil { 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 or a switch to inline. p.ready() - t := time.Now() - stall := p.handle(pkts, fwPacket, rejectBuf, conntrackCache.Get()) - p.readBusy.Inc(int64(time.Since(t) - stall)) + p.handle(pkts, fwPacket, rejectBuf, conntrackCache.Get()) } } // handle takes one tun read through the pipeline. It returns once the batch is queued, or sent when the reader sent -// 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 { +// it itself. +func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, rejectBuf []byte, localCache firewall.ConntrackCache) { p.ready() - inline, stall := p.gate(p.small(pkts)) + inline := 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 stall + return } slot := p.free[len(p.free)-1] @@ -373,23 +364,21 @@ func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, re if len(items) == 0 { slot.items = items p.free = append(p.free, slot) - return stall + return } if p.oversized() { p.metricInline.Inc(1) - stall += p.sendLarge(items) + p.sendLarge(items) slot.items = items[:0] p.free = append(p.free, slot) - return stall + return } - d, ok := p.waitBudget() - stall += d - if !ok { + if !p.waitBudget() { clear(items) slot.items = items[:0] p.free = append(p.free, slot) - return stall + return } slot.reservations = append(slot.reservations[:0], p.tunnels...) items = p.reserve(items) @@ -397,13 +386,13 @@ func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, re if len(items) == 0 { slot.releaseBudget() p.free = append(p.free, slot) - return stall + return } if inline { p.inline(slot) p.free = append(p.free, slot) - return stall + return } // The packets now live in the slot, and the queue reads into the slot's old buffer @@ -415,32 +404,30 @@ func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, re // before any other batch can, so the sender puts them on the wire in counter order. p.ordered <- slot p.pool.jobs <- slot - return stall } -// gate reports whether a read goes inline, and how long it waited for the pipeline to drain. Each routine holds one +// gate reports whether a read goes inline. 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) { +func (p *stagedTx) gate(small bool) bool { if small == p.gateInline { p.gateStreak = 0 - return small, 0 + return small } if p.gateStreak++; p.gateStreak < p.f.staged.gateHold { - return p.gateInline, 0 + return p.gateInline } p.gateInline, p.gateStreak = small, 0 p.metricGateSwitches.Inc(1) if !small { - return false, 0 + return false } - t := time.Now() for p.flushOldest() { } - return true, time.Since(t) + return true } // ready takes back every slot the sender has finished with, and waits for one when all of them are in flight. @@ -486,7 +473,6 @@ func (p *stagedTx) sender() { } for s := range p.ordered { <-s.done - t := time.Now() if p.f.closed.Load() { // The sockets are gone, a flush would only log errors s.sb.Reset() @@ -494,7 +480,6 @@ func (p *stagedTx) sender() { p.f.flushSendBatch(s.sb, p.q) } s.releaseBudget() - p.sendBusy.Inc(int64(time.Since(t))) clear(s.items) s.items = s.items[:0] p.returned <- s