tun: Read the clock once per staged loop iteration

The TX reader timed its slot wait with a second pair of clock reads only to
subtract it again. It now waits for a free slot before it starts the busy
clock, so a read costs one time.Now and one time.Since. The RX listener
likewise ends its busy time when it has to wait for a batch instead of
timing the wait, and pool workers time a run of queued jobs with one pair
of reads. A test checks the read stages still leave backpressure out.

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.

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 {

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")
+}

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.