From 49de171ad0e858154fd9c987c968cbe2b4949879 Mon Sep 17 00:00:00 2001 From: flamingcow Date: Tue, 4 Aug 2026 21:25:04 -0700 Subject: [PATCH] Commit the receive counters once per batch instead of once per frame --- main_test.go | 30 ++++++++++++++++++++---------- rx.go | 46 +++++++++++++++++++++++++++++++++++++++------- 2 files changed, 59 insertions(+), 17 deletions(-) diff --git a/main_test.go b/main_test.go index 2afa603..e0abb39 100644 --- a/main_test.go +++ b/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) - s.observe(5*int64(time.Millisecond), 200) - s.observe(rateBucketNs+int64(time.Millisecond), 300) + run := rateRun{stats: &s} + run.add(3*int64(time.Millisecond), 100) + 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) - s.observe(rateBuckets*rateBucketNs+1, 200) + run := rateRun{stats: &s} + 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) - d.rxStats[1].observe(rateBucketNs+2, 700) - d.rxStats[0].observe(2*rateBucketNs+1, 900) + r0 := rateRun{stats: d.rxStats[0]} + r1 := rateRun{stats: d.rxStats[1]} + 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) - d.rxStats[0].observe(2*rateBucketNs+1, 900) + run := rateRun{stats: d.rxStats[0]} + run.add(rateBucketNs+1, 500) + run.add(2*rateBucketNs+1, 900) + run.flush() now := time.Now() d.readRateBucket(now) diff --git a/rx.go b/rx.go index 9f9e6c2..b347362 100644 --- a/rx.go +++ b/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) { - e := stamp / rateBucketNs +func (s *rxStats) commit(e int64, frames, bytes uint64) { 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.bytes.Add(n) + b.frames.Add(frames) + 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) - w.stats.bytes.Add(uint64(len(buf))) + frames++ + 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() } }