Commit the receive counters once per batch instead of once per frame

main_test.go

@@ -9,9 +9,11 @@ import (
 // got round to draining it.
 func TestRxObserveBucketsByStamp(t *testing.T) {
 	var s rxStats
-	s.observe(3*int64(time.Millisecond), 100)
+	run := rateRun{stats: &s}
-	s.observe(5*int64(time.Millisecond), 200)
+	run.add(3*int64(time.Millisecond), 100)
-	s.observe(rateBucketNs+int64(time.Millisecond), 300)
+	run.add(5*int64(time.Millisecond), 200)
+	run.add(rateBucketNs+int64(time.Millisecond), 300)
+	run.flush()
 
 	if f, b := s.bucket(0); f != 2 || b != 300 {
 		t.Errorf("epoch 0 = %d frames, %d bytes; want 2, 300", f, b)
@@ -28,8 +30,10 @@ func TestRxObserveBucketsByStamp(t *testing.T) {
 // reports nothing rather than the stale counts.
 func TestRxBucketWraps(t *testing.T) {
 	var s rxStats
-	s.observe(1, 100)
+	run := rateRun{stats: &s}
-	s.observe(rateBuckets*rateBucketNs+1, 200)
+	run.add(1, 100)
+	run.add(rateBuckets*rateBucketNs+1, 200)
+	run.flush()
 
 	if f, b := s.bucket(0); f != 0 || b != 0 {
 		t.Errorf("evicted epoch = %d frames, %d bytes; want 0, 0", f, b)
@@ -42,9 +46,13 @@ func TestRxBucketWraps(t *testing.T) {
 // The newest epoch may still be filling, so the rate comes from the one before.
 func TestDirectionRateReadsOneBucketBack(t *testing.T) {
 	d := &direction{rxStats: []*rxStats{{}, {}}}
-	d.rxStats[0].observe(rateBucketNs+1, 500)
+	r0 := rateRun{stats: d.rxStats[0]}
-	d.rxStats[1].observe(rateBucketNs+2, 700)
+	r1 := rateRun{stats: d.rxStats[1]}
-	d.rxStats[0].observe(2*rateBucketNs+1, 900)
+	r0.add(rateBucketNs+1, 500)
+	r1.add(rateBucketNs+2, 700)
+	r0.add(2*rateBucketNs+1, 900)
+	r0.flush()
+	r1.flush()
 
 	d.readRateBucket(time.Now())
 	if d.rateFrames != 2 || d.rateBytes != 1200 {
@@ -57,8 +65,10 @@ func TestDirectionRateReadsOneBucketBack(t *testing.T) {
 // a quiet wire and must not keep reporting the last bucket.
 func TestDirectionRateGoesStale(t *testing.T) {
 	d := &direction{rxStats: []*rxStats{{}}}
-	d.rxStats[0].observe(rateBucketNs+1, 500)
+	run := rateRun{stats: d.rxStats[0]}
-	d.rxStats[0].observe(2*rateBucketNs+1, 900)
+	run.add(rateBucketNs+1, 500)
+	run.add(2*rateBucketNs+1, 900)
+	run.flush()
 
 	now := time.Now()
 	d.readRateBucket(now)

rx.go

@@ -39,21 +39,46 @@ type rxBucket struct {
 
 // Only the owning worker writes its own buckets, so a slot coming round again is
 // simply zeroed before it is claimed for the new epoch.
-func (s *rxStats) observe(stamp int64, n uint64) {
+func (s *rxStats) commit(e int64, frames, bytes uint64) {
-	e := stamp / rateBucketNs
 	b := &s.buckets[e&(rateBuckets-1)]
 	if b.epoch.Load() != e {
 		b.frames.Store(0)
 		b.bytes.Store(0)
 		b.epoch.Store(e)
 	}
-	b.frames.Add(1)
+	b.frames.Add(frames)
-	b.bytes.Add(n)
+	b.bytes.Add(bytes)
 	if e > s.newest.Load() {
 		s.newest.Store(e)
 	}
 }
 
+// Frames drained together that fell in the same epoch, so the buckets take one
+// pair of adds per epoch a batch spans rather than one per frame.
+type rateRun struct {
+	stats  *rxStats
+	epoch  int64
+	frames uint64
+	bytes  uint64
+}
+
+func (r *rateRun) add(stamp int64, n uint64) {
+	if e := stamp / rateBucketNs; e != r.epoch {
+		r.flush()
+		r.epoch = e
+	}
+	r.frames++
+	r.bytes += n
+}
+
+func (r *rateRun) flush() {
+	if r.frames == 0 {
+		return
+	}
+	r.stats.commit(r.epoch, r.frames, r.bytes)
+	r.frames, r.bytes = 0, 0
+}
+
 // What this worker counted into one epoch, or nothing if that epoch has already
 // fallen out of the ring.
 func (s *rxStats) bucket(e int64) (frames, bytes uint64) {
@@ -83,6 +108,7 @@ func (w *rxWorker) run(done *atomic.Bool) {
 		}
 	}
 	hdrs, oob := newRxMmsghdrs(bufs)
+	run := rateRun{stats: w.stats}
 
 	w.ready.Done()
 
@@ -99,6 +125,7 @@ func (w *rxWorker) run(done *atomic.Bool) {
 			}
 			continue
 		}
+		var frames, bytes uint64
 		for i := 0; i < n; i++ {
 			buf := bufs[i][:int(hdrs[i].len)]
 			p, st := parseHeader(buf)
@@ -110,10 +137,10 @@ func (w *rxWorker) run(done *atomic.Bool) {
 				}
 				continue
 			}
-			w.stats.frames.Add(1)
+			frames++
-			w.stats.bytes.Add(uint64(len(buf)))
+			bytes += uint64(len(buf))
 			if ts, ok := hwTimestamp(oob[i][:hdrs[i].hdr.Controllen]); ok {
-				w.stats.observe(ts, uint64(len(buf)))
+				run.add(ts, uint64(len(buf)))
 			}
 
 			// The ethertype this socket is bound to already says which stream the
@@ -134,5 +161,10 @@ func (w *rxWorker) run(done *atomic.Bool) {
 				w.stats.crcErr.Add(1)
 			}
 		}
+		// Nothing reads these between frames, and they share a cache line, so a
+		// batch commits once rather than locking the line for every frame.
+		w.stats.frames.Add(frames)
+		w.stats.bytes.Add(bytes)
+		run.flush()
 	}
 }