diff --git a/main.go b/main.go index 3ba02be..290c922 100644 --- a/main.go +++ b/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 { - n := w.count() - if n < 2 { - return 0 - } - 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) -} +// How long the newest epoch may sit still before the wire is taken to have gone +// quiet. A stamp only advances when a frame arrives, so a frozen epoch means no +// traffic rather than an unchanged rate. +const rateStale = 100 * time.Millisecond -func rxRatePPS(p, c counterSet, secs float64) float64 { - return float64(c.s.rxFrames-p.s.rxFrames) / secs -} - -func rxRateGbps(p, c counterSet, secs float64) float64 { - return gbps(c.s.rxBytes-p.s.rxBytes, c.s.rxFrames-p.s.rxFrames, secs) +// The newest epoch is still filling, since frames received in it may not have +// been drained yet, so the rate is read from the one before it. +func (d *direction) readRateBucket(now time.Time) { + var newest int64 + for _, r := range d.rxStats { + 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.rxGbps = d.win.latest(rxRateGbps) + v.rxPPS = float64(d.rateFrames) / rateBucketSecs + 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{}) } diff --git a/main_test.go b/main_test.go index 1f009ec..657e379 100644 --- a/main_test.go +++ b/main_test.go @@ -5,40 +5,87 @@ import ( "time" ) -func TestRateWindowLatestNeedsTwoBuckets(t *testing.T) { - w := newRateWindow(4) - if got := w.latest(rxRatePPS); got != 0 { - t.Errorf("empty ring gave %v, want 0", got) +// A frame lands in the bucket its receive stamp falls in, whenever the worker +// 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) + + 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 got := w.latest(rxRatePPS); got != 0 { - t.Errorf("one bucket gave %v, want 0", got) + if f, b := s.bucket(1); f != 1 || b != 300 { + t.Errorf("epoch 1 = %d frames, %d bytes; want 1, 300", f, b) + } + 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 -// averaged against everything still in the ring. -func TestRateWindowLatestUsesNewestPair(t *testing.T) { - w := newRateWindow(4) - t0 := time.Now() - w.push(counterSet{t: t0, s: sample{rxFrames: 100}}) - w.push(counterSet{t: t0.Add(time.Second), s: sample{rxFrames: 300}}) - if got := w.latest(rxRatePPS); got != 200 { - t.Errorf("rate = %v, want 200", got) +// A slot coming round again belongs to its new epoch, and the epoch it replaced +// 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) + + if f, b := s.bucket(0); f != 0 || b != 0 { + 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 got := w.latest(rxRatePPS); got != 100 { - t.Errorf("rate = %v, want 100 rather than the mean of the ring", got) + if f, b := s.bucket(rateBuckets); f != 1 || b != 200 { + t.Errorf("new epoch = %d frames, %d bytes; want 1, 200", f, b) } } -func TestRateWindowLatestIgnoresZeroSpan(t *testing.T) { - w := newRateWindow(4) - t0 := time.Now() - w.push(counterSet{t: t0, s: sample{rxFrames: 100}}) - w.push(counterSet{t: t0, s: sample{rxFrames: 300}}) - if got := w.latest(rxRatePPS); got != 0 { - t.Errorf("rate = %v, want 0", got) +// 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) + + 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) } } diff --git a/rx.go b/rx.go index fd11cdd..3149d59 100644 --- a/rx.go +++ b/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) diff --git a/sock.go b/sock.go index 79a9bc1..1acaf72 100644 --- a/sock.go +++ b/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 {