diff --git a/flow.go b/flow.go index 5425a4c5..92e22407 100644 --- a/flow.go +++ b/flow.go @@ -108,8 +108,12 @@ func newFlowPool(workers, queueLen int, busy metrics.Counter) *flowPool { 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() - job.run(nb) + for job != nil { + job.run(nb) + job = p.next() + } busy.Inc(int64(time.Since(t))) } }) @@ -117,6 +121,16 @@ func newFlowPool(workers, queueLen int, busy metrics.Counter) *flowPool { 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. diff --git a/flow_rx.go b/flow_rx.go index ee1b34a3..e0b5100c 100644 --- a/flow_rx.go +++ b/flow_rx.go @@ -209,10 +209,7 @@ func (p *stagedRx) precheck(b *rxBatch) { func (p *stagedRx) handoff(bufs [][]byte) [][]byte { p.cur.bufs = bufs p.endBatch() - if !p.started.IsZero() { - p.readBusy.Inc(int64(time.Since(p.started))) - p.started = time.Time{} - } + p.stopClock() b := p.cur if len(b.bufs) != len(bufs) || len(b.bufs[0]) < len(bufs[0]) { b.bufs = make([][]byte, len(bufs)) @@ -262,15 +259,23 @@ func (p *stagedRx) endBatch() { 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, not listener busy time. - t := time.Now() + // downstream backpressure, so the listener's busy time ends here. + p.stopClock() p.release(<-p.returned) - p.started = p.started.Add(time.Since(t)) } 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 { diff --git a/flow_test.go b/flow_test.go index 25e90bc0..4eb67a2e 100644 --- a/flow_test.go +++ b/flow_test.go @@ -533,3 +533,34 @@ 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 cd479fe8..0df643bf 100644 --- a/flow_tx.go +++ b/flow_tx.go @@ -338,6 +338,9 @@ 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. + p.ready() t := time.Now() stall := p.handle(pkts, fwPacket, rejectBuf, conntrackCache.Get()) p.readBusy.Inc(int64(time.Since(t) - stall)) @@ -345,9 +348,9 @@ func (p *stagedTx) run(fwPacket *firewall.ParsedPacket, rejectBuf []byte, conntr } // 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 the sender. +// 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.reap() + p.ready() 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 @@ -356,7 +359,9 @@ func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, re return 0 } - slot, stall := p.acquire() + 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 @@ -406,18 +411,12 @@ func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, re return stall } -// acquire returns an idle slot, waiting for the sender to give one back when all of them are in flight, and how long -// it waited. -func (p *stagedTx) acquire() (*txSlot, time.Duration) { - var stall time.Duration +// 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() if len(p.free) == 0 { - t := time.Now() p.release(<-p.returned) - stall = time.Since(t) } - s := p.free[len(p.free)-1] - p.free = p.free[:len(p.free)-1] - return s, stall } // reap takes back every slot the sender has finished with, without waiting.