Bucket received frames by their mac receive stamp and read one bucket back

main.go

@@ -53,6 +53,14 @@ type direction struct {
 	drops uint64
 	base  counterSet
 
+	// The completed receive bucket the display draws its rate from, refreshed by
+	// the sampler because the buckets are keyed by the mac's clock and staleness
+	// has to be judged against the wall.
+	rateFrames uint64
+	rateBytes  uint64
+	epoch      int64
+	epochAt    time.Time
+
 	nic    atomic.Uint64
 	poller *nicPoller
 }
@@ -129,25 +137,32 @@ func (w *rateWindow) at(i int) counterSet {
 	return w.buf[i%len(w.buf)]
 }
 
-func (w *rateWindow) latest(rate func(prev, cur counterSet, secs float64) float64) float64 {
+// How long the newest epoch may sit still before the wire is taken to have gone
-	n := w.count()
+// quiet. A stamp only advances when a frame arrives, so a frozen epoch means no
-	if n < 2 {
+// traffic rather than an unchanged rate.
-		return 0
+const rateStale = 100 * time.Millisecond
-	}
-	prev, cur := w.at(n-2), w.at(n-1)
-	secs := cur.t.Sub(prev.t).Seconds()
-	if secs <= 0 {
-		return 0
-	}
-	return rate(prev, cur, secs)
-}
 
-func rxRatePPS(p, c counterSet, secs float64) float64 {
+// The newest epoch is still filling, since frames received in it may not have
-	return float64(c.s.rxFrames-p.s.rxFrames) / secs
+// been drained yet, so the rate is read from the one before it.
-}
+func (d *direction) readRateBucket(now time.Time) {
-
+	var newest int64
-func rxRateGbps(p, c counterSet, secs float64) float64 {
+	for _, r := range d.rxStats {
-	return gbps(c.s.rxBytes-p.s.rxBytes, c.s.rxFrames-p.s.rxFrames, secs)
+		if e := r.newest.Load(); e > newest {
+			newest = e
+		}
+	}
+	if newest > d.epoch {
+		d.epoch, d.epochAt = newest, now
+	}
+	d.rateFrames, d.rateBytes = 0, 0
+	if d.epoch == 0 || now.Sub(d.epochAt) > rateStale {
+		return
+	}
+	for _, r := range d.rxStats {
+		f, b := r.bucket(d.epoch - 1)
+		d.rateFrames += f
+		d.rateBytes += b
+	}
 }
 
 type sample struct {
@@ -304,6 +319,7 @@ func totalView(views []view) view {
 func (d *direction) sample() {
 	d.mu.Lock()
 	d.win.push(d.capture())
+	d.readRateBucket(time.Now())
 	d.mu.Unlock()
 }
 
@@ -320,8 +336,8 @@ func (d *direction) displayView() view {
 	if n >= 2 {
 		v.window = errsBetween(d.win.at(0), d.win.at(n-1))
 	}
-	v.rxPPS = d.win.latest(rxRatePPS)
+	v.rxPPS = float64(d.rateFrames) / rateBucketSecs
-	v.rxGbps = d.win.latest(rxRateGbps)
+	v.rxGbps = gbps(d.rateBytes, d.rateFrames, rateBucketSecs)
 	d.mu.Unlock()
 	return v
 }
@@ -389,6 +405,11 @@ func buildDirection(label string, tx, rx endpoint) (*direction, error) {
 		if err != nil {
 			return nil, fmt.Errorf("%s rx socket for 0x%04x: %w", label, et, err)
 		}
+		// The mac already stamps every frame for the probe's sake, so this only
+		// asks for the stamp to be delivered.
+		if err := enableRxTimestamps(fd); err != nil {
+			return nil, fmt.Errorf("%s rx timestamps for 0x%04x: %w", label, et, err)
+		}
 		d.rxFDs = append(d.rxFDs, fd)
 		d.rxStats = append(d.rxStats, &rxStats{})
 	}

main_test.go

@@ -5,40 +5,87 @@ import (
 	"time"
 )
 
-func TestRateWindowLatestNeedsTwoBuckets(t *testing.T) {
+// A frame lands in the bucket its receive stamp falls in, whenever the worker
-	w := newRateWindow(4)
+// got round to draining it.
-	if got := w.latest(rxRatePPS); got != 0 {
+func TestRxObserveBucketsByStamp(t *testing.T) {
-		t.Errorf("empty ring gave %v, want 0", got)
+	var s rxStats
+	s.observe(3*int64(time.Millisecond), 100)
+	s.observe(5*int64(time.Millisecond), 200)
+	s.observe(rateBucketNs+int64(time.Millisecond), 300)
+
+	if f, b := s.bucket(0); f != 2 || b != 300 {
+		t.Errorf("epoch 0 = %d frames, %d bytes; want 2, 300", f, b)
 	}
-	w.push(counterSet{t: time.Now(), s: sample{rxFrames: 100}})
+	if f, b := s.bucket(1); f != 1 || b != 300 {
-	if got := w.latest(rxRatePPS); got != 0 {
+		t.Errorf("epoch 1 = %d frames, %d bytes; want 1, 300", f, b)
-		t.Errorf("one bucket gave %v, want 0", got)
+	}
+	if got := s.newest.Load(); got != 1 {
+		t.Errorf("newest = %d, want 1", got)
 	}
 }
 
-// The newest pair alone, so a step in the rate shows at once instead of being
+// A slot coming round again belongs to its new epoch, and the epoch it replaced
-// averaged against everything still in the ring.
+// reports nothing rather than the stale counts.
-func TestRateWindowLatestUsesNewestPair(t *testing.T) {
+func TestRxBucketWraps(t *testing.T) {
-	w := newRateWindow(4)
+	var s rxStats
-	t0 := time.Now()
+	s.observe(1, 100)
-	w.push(counterSet{t: t0, s: sample{rxFrames: 100}})
+	s.observe(rateBuckets*rateBucketNs+1, 200)
-	w.push(counterSet{t: t0.Add(time.Second), s: sample{rxFrames: 300}})
+
-	if got := w.latest(rxRatePPS); got != 200 {
+	if f, b := s.bucket(0); f != 0 || b != 0 {
-		t.Errorf("rate = %v, want 200", got)
+		t.Errorf("evicted epoch = %d frames, %d bytes; want 0, 0", f, b)
 	}
-	w.push(counterSet{t: t0.Add(2 * time.Second), s: sample{rxFrames: 400}})
+	if f, b := s.bucket(rateBuckets); f != 1 || b != 200 {
-	if got := w.latest(rxRatePPS); got != 100 {
+		t.Errorf("new epoch = %d frames, %d bytes; want 1, 200", f, b)
-		t.Errorf("rate = %v, want 100 rather than the mean of the ring", got)
 	}
 }
 
-func TestRateWindowLatestIgnoresZeroSpan(t *testing.T) {
+// The newest epoch may still be filling, so the rate comes from the one before.
-	w := newRateWindow(4)
+func TestDirectionRateReadsOneBucketBack(t *testing.T) {
-	t0 := time.Now()
+	d := &direction{rxStats: []*rxStats{{}, {}}}
-	w.push(counterSet{t: t0, s: sample{rxFrames: 100}})
+	d.rxStats[0].observe(rateBucketNs+1, 500)
-	w.push(counterSet{t: t0, s: sample{rxFrames: 300}})
+	d.rxStats[1].observe(rateBucketNs+2, 700)
-	if got := w.latest(rxRatePPS); got != 0 {
+	d.rxStats[0].observe(2*rateBucketNs+1, 900)
-		t.Errorf("rate = %v, want 0", got)
+
+	d.readRateBucket(time.Now())
+	if d.rateFrames != 2 || d.rateBytes != 1200 {
+		t.Errorf("rate bucket = %d frames, %d bytes; want the completed epoch, 2 and 1200",
+			d.rateFrames, d.rateBytes)
+	}
+}
+
+// A stamp only advances when a frame arrives, so an epoch that stops moving is
+// 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)
+
+	now := time.Now()
+	d.readRateBucket(now)
+	if d.rateFrames == 0 {
+		t.Fatal("expected a rate while the epoch was still moving")
+	}
+	d.readRateBucket(now.Add(2 * rateStale))
+	if d.rateFrames != 0 || d.rateBytes != 0 {
+		t.Errorf("stale rate = %d frames, %d bytes; want 0, 0", d.rateFrames, d.rateBytes)
+	}
+}
+
+// Indexed oldest first, so the error window still spans the whole ring once it
+// has wrapped.
+func TestRateWindowIndexesOldestFirst(t *testing.T) {
+	w := newRateWindow(3)
+	for i := 1; i <= 5; i++ {
+		w.push(counterSet{s: sample{rxFrames: uint64(i)}})
+	}
+	if got := w.count(); got != 3 {
+		t.Fatalf("count = %d, want 3", got)
+	}
+	if got := w.at(0).s.rxFrames; got != 3 {
+		t.Errorf("oldest = %d, want 3", got)
+	}
+	if got := w.at(2).s.rxFrames; got != 5 {
+		t.Errorf("newest = %d, want 5", got)
 	}
 }
 

rx.go

@@ -15,6 +15,52 @@ type rxStats struct {
 	badLen   atomic.Uint64
 	crcErr   atomic.Uint64
 	rxErrs   atomic.Uint64
+
+	// Frames counted into the interval their mac receive stamp falls in, rather
+	// than the interval a worker got round to draining them in.
+	newest  atomic.Int64
+	buckets [rateBuckets]rxBucket
+}
+
+// One sample interval of arrivals, keyed by the mac's clock, with enough of them
+// kept that a bucket is read long before its slot comes round again.
+const (
+	rateBucketNs   = int64(sampleInterval)
+	rateBuckets    = 64
+	rateBucketSecs = float64(rateBucketNs) / 1e9
+)
+
+type rxBucket struct {
+	epoch  atomic.Int64
+	frames atomic.Uint64
+	bytes  atomic.Uint64
+}
+
+// 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
+	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)
+	if e > s.newest.Load() {
+		s.newest.Store(e)
+	}
+}
+
+// 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) {
+	b := &s.buckets[e&(rateBuckets-1)]
+	if b.epoch.Load() != e {
+		return 0, 0
+	}
+	return b.frames.Load(), b.bytes.Load()
 }
 
 type rxWorker struct {
@@ -34,11 +80,16 @@ func (w *rxWorker) run(done *atomic.Bool) {
 			bufs[i][j] = 0
 		}
 	}
-	hdrs, _ := newMmsghdrs(bufs)
+	hdrs, oob := newRxMmsghdrs(bufs)
 
 	w.ready.Done()
 
 	for !done.Load() {
+		// The kernel overwrites each Controllen with what it wrote, so they are
+		// reset before every call.
+		for i := range hdrs {
+			hdrs[i].hdr.Controllen = cmsgLen
+		}
 		n, err := recvmmsg(w.fd, hdrs, unix.MSG_WAITFORONE)
 		if n <= 0 {
 			if err != nil && err != unix.EAGAIN && err != unix.EINTR {
@@ -55,6 +106,9 @@ func (w *rxWorker) run(done *atomic.Bool) {
 			}
 			w.stats.frames.Add(1)
 			w.stats.bytes.Add(uint64(len(buf)))
+			if ts, ok := hwTimestamp(oob[i][:hdrs[i].hdr.Controllen]); ok {
+				w.stats.observe(ts, uint64(len(buf)))
+			}
 
 			if int(p.stream) < len(w.streams) {
 				w.streams[p.stream].observe(p.seq)

sock.go

@@ -114,6 +114,22 @@ func newMmsghdrs(bufs [][]byte) ([]mmsghdr, []unix.Iovec) {
 	return hdrs, iovs
 }
 
+// Room for one SCM_TIMESTAMPING and its three timespecs.
+const cmsgLen = 128
+
+// Receive headers carry a control buffer each, so the mac's receive stamp comes
+// back alongside every frame.
+func newRxMmsghdrs(bufs [][]byte) ([]mmsghdr, [][]byte) {
+	hdrs, _ := newMmsghdrs(bufs)
+	oob := make([][]byte, len(bufs))
+	for i := range bufs {
+		oob[i] = make([]byte, cmsgLen)
+		hdrs[i].hdr.Control = &oob[i][0]
+		hdrs[i].hdr.Controllen = cmsgLen
+	}
+	return hdrs, oob
+}
+
 func packetDrops(fd int) uint64 {
 	st, err := unix.GetsockoptTpacketStats(fd, unix.SOL_PACKET, unix.PACKET_STATISTICS)
 	if err != nil {