Schedule GRO flush iff packets are queued

Originally Ghanan's cl/494241863. This prevents goroutines from being unnecessarily scheduled when there are no packets in the GRO dispatcher, reducing wasted CPU cycles. With this change, a timer to flush GRO packets is scheduled when a packet is enqueued and no previous timer exists. The timer is rescheduled to be fired again if after flushing packets, packets remain in the GRO dispatcher. If GRO is disabled, and a timer was previously set, the timer is reset to fire immediately and flush all packets. Performance: The big win here is for non-network-heavy workloads. The GRO flush thread no longer fires when unnecessary. GRO used to add absurd overhead for such workloads. Now it adds none: e.g. `sleep 10` used <= 10ms without GRO, 7.79s with it (!!!), and now uses <= 10ms with this change. This lowers throughput and CPU usage roughly proportionally in networking-heavy benchmarking. CPU usage seems to decrease slightly more than throughput (which is good!). PiperOrigin-RevId: 516720406
2026-05-22 17:12:49 -07:00 · 2023-03-14 21:51:54 -07:00
parent 897c03039e
commit 9abcf28e60
1 changed files with 69 additions and 57 deletions
@@ -24,6 +24,7 @@ import (
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 )

+// TODO(b/256037250): Enable by default.
 // TODO(b/256037250): We parse headers here. We should save those headers in
 // PacketBuffers so they don't have to be re-parsed later.
 // TODO(b/256037250): I still see the occasional SACK block in the zero-loss
@@ -33,7 +34,6 @@ import (
 // opportunity for coalescing.
 // TODO(b/256037250): We're doing some header parsing here, which presents the
 // opportunity to skip it later.
-// TODO(b/256037250): Disarm or ignore the timer when GRO is empty.
 // TODO(b/256037250): We may be able to remove locking by pairing
 // groDispatchers with link endpoint dispatchers.

@@ -222,7 +222,7 @@ func (gb *groBucket) findGROPacket6(pkt PacketBufferPtr, ipHdr header.IPv6, tcpH
 }

 // +checklocks:gb.mu
-func (gb *groBucket) found(groPkt *groPacket, flushGROPkt bool, pkt PacketBufferPtr, ipHdr []byte, tcpHdr header.TCP, ep NetworkEndpoint, updateIPHdr func([]byte, int)) {
+func (gb *groBucket) found(gd *groDispatcher, groPkt *groPacket, flushGROPkt bool, pkt PacketBufferPtr, ipHdr []byte, tcpHdr header.TCP, ep NetworkEndpoint, updateIPHdr func([]byte, int)) {
 	// Flush groPkt or merge the packets.
 	pktSize := pkt.Data().Size()
 	flags := tcpHdr.Flags()
@@ -294,6 +294,11 @@ func (gb *groBucket) found(groPkt *groPacket, flushGROPkt bool, pkt PacketBuffer
 		// A merge occurred and we don't need to flush anything.
 		gb.mu.Unlock()
 	}
+
+	// Schedule a timer if we never had one set before.
+	if gd.flushTimerState.CompareAndSwap(flushTimerUnset, flushTimerSet) {
+		gd.flushTimer.Reset(gd.getInterval())
+	}
 }

 // A groPacket is packet undergoing GRO. It may be several packets coalesced
@@ -341,23 +346,26 @@ func (pk *groPacket) payloadSize() int {
 	return pk.pkt.Data().Size() - len(pk.ipHdr) - int(pk.tcpHdr.DataOffset())
 }

+// Values held in groDispatcher.flushTimerState.
+const (
+	flushTimerUnset = iota
+	flushTimerSet
+	flushTimerClosed
+)
+
 // groDispatcher coalesces incoming packets to increase throughput.
 type groDispatcher struct {
-	// newInterval notifies about changes to the interval.
-	newInterval chan struct{}
 	// intervalNS is the interval in nanoseconds.
 	intervalNS atomicbitops.Int64
-	// stop instructs the GRO dispatcher goroutine to stop.
-	stop chan struct{}

 	buckets [groNBuckets]groBucket
-	wg      sync.WaitGroup
+
+	flushTimerState atomicbitops.Int32
+	flushTimer      *time.Timer
 }

 func (gd *groDispatcher) init(interval time.Duration) {
 	gd.intervalNS.Store(interval.Nanoseconds())
-	gd.newInterval = make(chan struct{}, 1)
-	gd.stop = make(chan struct{})

 	for i := range gd.buckets {
 		bucket := &gd.buckets[i]
@@ -369,59 +377,49 @@ func (gd *groDispatcher) init(interval time.Duration) {
 		bucket.mu.Unlock()
 	}

-	gd.start(interval)
-}
+	// Create a timer to fire far from now and cancel it immediately.
+	//
+	// The timer will be reset when there is a need for it to fire.
+	gd.flushTimer = time.AfterFunc(time.Hour, func() {
+		if !gd.flushTimerState.CompareAndSwap(flushTimerSet, flushTimerUnset) {
+			// Timer was unset or GRO is closed, do nothing further.
+			return
+		}

-// start spawns a goroutine that flushes the GRO periodically based on the
-// interval.
-func (gd *groDispatcher) start(interval time.Duration) {
-	gd.wg.Add(1)
-
-	go func(interval time.Duration) {
-		defer gd.wg.Done()
-
-		var ch <-chan time.Time
+		interval := gd.getInterval()
 		if interval == 0 {
-			// Never run.
-			ch = make(<-chan time.Time)
-		} else {
-			ticker := time.NewTicker(interval)
-			ch = ticker.C
+			gd.flushAll()
+			return
 		}
-		for {
-			select {
-			case <-gd.newInterval:
-				interval = time.Duration(gd.intervalNS.Load()) * time.Nanosecond
-				if interval == 0 {
-					// Never run. Flush any existing GRO packets.
-					gd.flushAll()
-					ch = make(<-chan time.Time)
-				} else {
-					ticker := time.NewTicker(interval)
-					ch = ticker.C
-				}
-			case <-ch:
-				gd.flush()
-			case <-gd.stop:
-				return
-			}
+
+		if gd.flush() && gd.flushTimerState.CompareAndSwap(flushTimerUnset, flushTimerSet) {
+			// Only reset the timer if we have more packets and the timer was
+			// previously unset. If we have no packets left, the timer is already set
+			// or GRO is being closed, do not reset the timer.
+			gd.flushTimer.Reset(interval)
 		}
-	}(interval)
+	})
+	gd.flushTimer.Stop()
 }

 func (gd *groDispatcher) getInterval() time.Duration {
 	return time.Duration(gd.intervalNS.Load()) * time.Nanosecond
 }

+// setInterval is not thread-safe and so much be protected by callers.
 func (gd *groDispatcher) setInterval(interval time.Duration) {
 	gd.intervalNS.Store(interval.Nanoseconds())
-	gd.newInterval <- struct{}{}
+
+	if gd.flushTimerState.Load() == flushTimerSet {
+		// Timer was previously set, reset it.
+		gd.flushTimer.Reset(interval)
+	}
 }

 // dispatch sends pkt up the stack after it undergoes GRO coalescing.
 func (gd *groDispatcher) dispatch(pkt PacketBufferPtr, netProto tcpip.NetworkProtocolNumber, ep NetworkEndpoint) {
 	// If GRO is disabled simply pass the packet along.
-	if gd.intervalNS.Load() == 0 {
+	if gd.getInterval() == 0 {
 		ep.HandlePacket(pkt)
 		return
 	}
@@ -503,7 +501,7 @@ func (gd *groDispatcher) dispatch4(pkt PacketBufferPtr, ep NetworkEndpoint) {
 	bucket := &gd.buckets[gd.bucketForPacket(ipHdr, tcpHdr)&groNBucketsMask]
 	bucket.mu.Lock()
 	groPkt, flushGROPkt := bucket.findGROPacket4(pkt, ipHdr, tcpHdr, ep)
-	bucket.found(groPkt, flushGROPkt, pkt, ipHdr, tcpHdr, ep, updateIPv4Hdr)
+	bucket.found(gd, groPkt, flushGROPkt, pkt, ipHdr, tcpHdr, ep, updateIPv4Hdr)
 }

 func (gd *groDispatcher) dispatch6(pkt PacketBufferPtr, ep NetworkEndpoint) {
@@ -601,7 +599,7 @@ func (gd *groDispatcher) dispatch6(pkt PacketBufferPtr, ep NetworkEndpoint) {
 	bucket := &gd.buckets[gd.bucketForPacket(ipHdr, tcpHdr)&groNBucketsMask]
 	bucket.mu.Lock()
 	groPkt, flushGROPkt := bucket.findGROPacket6(pkt, ipHdr, tcpHdr, ep)
-	bucket.found(groPkt, flushGROPkt, pkt, ipHdr, tcpHdr, ep, updateIPv6Hdr)
+	bucket.found(gd, groPkt, flushGROPkt, pkt, ipHdr, tcpHdr, ep, updateIPv6Hdr)
 }

 func (gd *groDispatcher) bucketForPacket(ipHdr header.Network, tcpHdr header.TCP) int {
@@ -620,18 +618,26 @@ func (gd *groDispatcher) bucketForPacket(ipHdr header.Network, tcpHdr header.TCP
 }

 // flush sends any packets older than interval up the stack.
-func (gd *groDispatcher) flush() {
+//
+// Returns true iff packets remain.
+func (gd *groDispatcher) flush() bool {
 	interval := gd.intervalNS.Load()
 	old := time.Now().Add(-time.Duration(interval) * time.Nanosecond)
-	gd.flushSince(old)
+	return gd.flushSinceOrEqualTo(old)
 }

-func (gd *groDispatcher) flushSince(old time.Time) {
+// flushSinceOrEqualTo sends any packets older than or equal to the specified
+// time.
+//
+// Returns true iff packets remain.
+func (gd *groDispatcher) flushSinceOrEqualTo(old time.Time) bool {
 	type pair struct {
 		pkt PacketBufferPtr
 		ep  NetworkEndpoint
 	}

+	hasMore := false
+
 	for i := range gd.buckets {
 		// Put packets in a slice so we don't have to hold bucket.mu
 		// when we call HandlePacket.
@@ -641,13 +647,14 @@ func (gd *groDispatcher) flushSince(old time.Time) {
 		bucket := &gd.buckets[i]
 		bucket.mu.Lock()
 		for groPkt := bucket.packets.Front(); groPkt != nil; groPkt = groPkt.Next() {
-			if groPkt.created.Before(old) {
-				pairs = append(pairs, pair{groPkt.pkt, groPkt.ep})
-				bucket.removeOne(groPkt)
-			} else {
+			if groPkt.created.After(old) {
 				// Packets are ordered by age, so we can move
 				// on once we find one that's too new.
+				hasMore = true
 				break
+			} else {
+				pairs = append(pairs, pair{groPkt.pkt, groPkt.ep})
+				bucket.removeOne(groPkt)
 			}
 		}
 		bucket.mu.Unlock()
@@ -657,16 +664,21 @@ func (gd *groDispatcher) flushSince(old time.Time) {
 			pair.pkt.DecRef()
 		}
 	}
+
+	return hasMore
 }

 func (gd *groDispatcher) flushAll() {
-	gd.flushSince(time.Now())
+	if gd.flushSinceOrEqualTo(time.Now()) {
+		panic("packets unexpectedly remain after flushing all")
+	}
 }

 // close stops the GRO goroutine and releases any held packets.
 func (gd *groDispatcher) close() {
-	gd.stop <- struct{}{}
-	gd.wg.Wait()
+	gd.flushTimer.Stop()
+	// Prevent the timer from being scheduled again.
+	gd.flushTimerState.Store(flushTimerClosed)

 	for i := range gd.buckets {
 		bucket := &gd.buckets[i]