cabletest/direction.go

package main

import (
	"fmt"
	"math"
	"sync"
	"sync/atomic"
	"time"

	"golang.org/x/sys/unix"
)

const wireOverhead = 24

type direction struct {
	specs   []*frameSpec
	txStats []*txStats
	rxStats []*rxStats
	streams []lossWindow
	txFDs   []int
	rxFDs   []int
	statFD  int

	// The mask gate reads this: while a cable measure runs, every fault delta
	// is dropped whole, so the measure's own link blip never reaches an
	// aggregate.
	masked *atomic.Bool

	// Guards everything the sampler touches. Collection consumes self-clearing
	// sources, so a second concurrent reader would see a gap.
	mu   sync.Mutex
	prev sample
	nic  *nicCounters

	// Since the reset: plain sums of the deltas each tick lets through. Frames
	// and bytes are traffic rather than faults and skip the mask gate.
	frames uint64
	bytes  uint64
	since  errs
	window *errsRing

	// The smeared receive bucket the display draws its rate from, refreshed by
	// the sampler because the buckets are keyed by the shared read clock and
	// staleness has to be judged against the wall.
	rateFrames uint64
	rateBytes  uint64
	epoch      int64
	epochAt    time.Time

	// The settled window: each completed bucket enters once, donates its
	// excess backward once, and leaves for display once no later bucket can
	// still refill it. Recomputing the smear from the raw ring every sample
	// instead would show each excess twice — once as the donation and again,
	// unspent, when its bucket reaches the display slot of a later window.
	smFrames [smearWindow]uint64
	smBytes  [smearWindow]uint64
	smLen    int
	smNext   int64
}

// What someone testing a cable is asking, rather than how each failure happened
// to be noticed.
type errs struct {
	lost     uint64
	corrupt  uint64
	link     uint64
	internal uint64
}

func (e errs) add(o errs) errs {
	return errs{
		lost: e.lost + o.lost, corrupt: e.corrupt + o.corrupt,
		link: e.link + o.link, internal: e.internal + o.internal,
	}
}

func (e errs) sub(o errs) errs {
	return errs{
		lost: e.lost - o.lost, corrupt: e.corrupt - o.corrupt,
		link: e.link - o.link, internal: e.internal - o.internal,
	}
}

// The rolling short-term window: one slot of accepted fault deltas per sampler
// tick, spanning rateWindowSpan, with the sum kept current as slots enter and
// leave. A masked tick pushes zero, so the window slides on the clock
// regardless.
type errsRing struct {
	slots []errs
	idx   int
	sum   errs
}

func newErrsRing(n int) *errsRing {
	return &errsRing{slots: make([]errs, n)}
}

func (r *errsRing) push(e errs) {
	r.sum = r.sum.sub(r.slots[r.idx]).add(e)
	r.slots[r.idx] = e
	r.idx++
	if r.idx == len(r.slots) {
		r.idx = 0
	}
}

// How long the frontier may sit still before the wire is taken to have gone
// quiet. It only advances when frames arrive on every stream, so a frozen
// frontier means a stream has stopped delivering rather than an unchanged rate.
const rateStale = 100 * time.Millisecond

// An epoch is only reachable once a worker drains frames into it, so each
// stream's newest epoch is a frontier: everything that queue has been read
// through. Epochs behind the lowest frontier — the leader's would claim
// buckets the stragglers are still filling — are closed to further commits,
// so each is settled into the window exactly once. The display takes the
// window's oldest bucket, the one no later bucket can still refill, so the
// headline runs one window behind real time.
func (d *direction) readRateBucket(now time.Time) {
	newest := int64(math.MaxInt64)
	for _, r := range d.rxStats {
		if e := r.newest.Load(); e < newest {
			newest = e
		}
	}
	if newest > d.epoch {
		d.epoch, d.epochAt = newest, now
	}
	if d.smNext == 0 && d.epoch > 1 {
		d.smNext = d.epoch - 1
	}
	for e := d.smNext; e > 0 && e < d.epoch; e++ {
		d.settle(e)
		d.smNext = e + 1
	}
	d.rateFrames, d.rateBytes = 0, 0
	if d.smLen == 0 || now.Sub(d.epochAt) > rateStale {
		return
	}
	d.rateFrames, d.rateBytes = d.smFrames[0], d.smBytes[0]
}

func (d *direction) settle(e int64) {
	if d.smLen == smearWindow {
		copy(d.smFrames[:], d.smFrames[1:])
		copy(d.smBytes[:], d.smBytes[1:])
		d.smLen--
	}
	var f, b uint64
	for _, r := range d.rxStats {
		rf, rb := r.bucket(e)
		f += rf
		b += rb
	}
	d.smFrames[d.smLen], d.smBytes[d.smLen] = f, b
	d.smLen++
	fillBack(d.smFrames[:d.smLen], d.smBytes[:d.smLen])
}

type sample struct {
	rxFrames, rxBytes uint64
	lost, late        uint64
	crcErr, badMagic  uint64
	badHdr            uint64
	badLen            uint64
	txErrs            uint64
	rxErrs            uint64
}

func (d *direction) snapshot() sample {
	var s sample
	for _, t := range d.txStats {
		s.txErrs += t.errs.Load()
	}
	for _, r := range d.rxStats {
		s.rxFrames += r.frames.Load()
		s.rxBytes += r.bytes.Load()
		s.crcErr += r.crcErr.Load()
		s.badMagic += r.badMagic.Load()
		s.badHdr += r.badHdr.Load()
		s.badLen += r.badLen.Load()
		s.rxErrs += r.rxErrs.Load()
	}
	for i := range d.streams {
		s.lost += d.streams[i].lost.Load()
		s.late += d.streams[i].late.Load()
	}
	return s
}

// Counters keep climbing in the workers; a reset zeroes the sums accumulated
// from them. Rates and the rolling window are about now rather than since the
// reset, so they keep running.
func (d *direction) reset() {
	d.mu.Lock()
	d.frames, d.bytes = 0, 0
	d.since = errs{}
	d.mu.Unlock()
}

func gbps(bytes, frames uint64, secs float64) float64 {
	return float64((bytes+frames*wireOverhead)*8) / secs / 1e9
}

// Shared by the console table and the framebuffer so both show the same
// figures.
type view struct {
	rxPPS             float64
	rxGbps            float64
	rxFrames, rxBytes uint64
	since             errs
	window            errs
}

// One tick of every counter this direction owns, flattened to deltas: the
// worker counters grow forever, the socket drop statistic clears itself on
// read, and the nic counters re-base, none of which survives past here.
func (d *direction) collect() (frames, bytes uint64, e errs) {
	cur := d.snapshot()
	prev := d.prev
	d.prev = cur
	var drops uint64
	for _, fd := range d.rxFDs {
		drops += packetDrops(fd)
	}
	return cur.rxFrames - prev.rxFrames, cur.rxBytes - prev.rxBytes,
		classify(cur, prev, drops, d.nic.delta())
}

// Which delta feeds which bucket is the whole taxonomy the panel reports.
func classify(cur, prev sample, drops, nic uint64) errs {
	return errs{
		lost: cur.lost - prev.lost,
		// Four ways of noticing one thing: a payload that does not match its
		// checksum, a header that does not match its own, a header that is not
		// ours, and a length that cannot be.
		corrupt: (cur.crcErr - prev.crcErr) + (cur.badHdr - prev.badHdr) +
			(cur.badMagic - prev.badMagic) + (cur.badLen - prev.badLen),
		// What the hardware reported. Nothing the host declined to send is here,
		// so this one going red means the cable.
		link: nic + (cur.rxErrs - prev.rxErrs),
		// Ours rather than the cable's. A late frame is unreachable while each
		// stream has a flow rule to its own queue, which is exactly why it is
		// worth counting.
		internal: drops + (cur.late - prev.late) + (cur.txErrs - prev.txErrs),
	}
}

// The mask gate: a masked tick is a cable measure's own blip, and its faults
// stop dead here — only the clock reaches the aggregates.
func (d *direction) aggregate(frames, bytes uint64, e errs) errs {
	d.frames += frames
	d.bytes += bytes
	if d.masked.Load() {
		e = errs{}
	}
	d.since = d.since.add(e)
	d.window.push(e)
	return e
}

func totalView(views []view) view {
	var t view
	for _, v := range views {
		t.rxPPS += v.rxPPS
		t.rxGbps += v.rxGbps
		t.rxFrames += v.rxFrames
		t.rxBytes += v.rxBytes
		t.since = t.since.add(v.since)
		t.window = t.window.add(v.window)
	}
	return t
}

// Returns the accepted deltas so the sampler can bucket them onward. The
// clock is read after the counters, per docs/measurement.md: stamps come from
// read time, never from the ticker.
func (d *direction) sample() errs {
	d.mu.Lock()
	defer d.mu.Unlock()
	frames, bytes, e := d.collect()
	e = d.aggregate(frames, bytes, e)
	d.readRateBucket(time.Now())
	return e
}

// Draws what the sampler last aggregated rather than reading the counters
// again, so the display never participates in the measurement.
func (d *direction) displayView() view {
	d.mu.Lock()
	v := view{
		rxFrames: d.frames,
		rxBytes:  d.bytes,
		since:    d.since,
		window:   d.window.sum,
		rxPPS:    float64(d.rateFrames) / rateBucketSecs,
		rxGbps:   gbps(d.rateBytes, d.rateFrames, rateBucketSecs),
	}
	d.mu.Unlock()
	return v
}

// Whatever was counted before now is not ours: the first collection primes
// every baseline and is discarded.
func (d *direction) primeCounters() {
	d.mu.Lock()
	d.collect()
	d.mu.Unlock()
}

func buildDirection(label string, tx, rx endpoint, masked *atomic.Bool) (*direction, error) {
	// Built before the windows, since each window judges sequence numbers against
	// the frontier its own sender publishes.
	txs := make([]*txStats, numStreams)
	for i := range txs {
		txs[i] = &txStats{}
	}
	d := &direction{
		txStats: txs,
		streams: newLossWindows(txs),
		masked:  masked,
	}
	// Held open for the life of the run: the stats ioctl rides every sampler
	// tick and reopening a socket for each one is pure overhead.
	statFD, err := unix.Socket(unix.AF_INET, unix.SOCK_DGRAM, 0)
	if err != nil {
		return nil, fmt.Errorf("%s stats socket: %w", label, err)
	}
	d.statFD = statFD

	d.nic, err = newNICCounters(statFD, tx.name, rx.name)
	if err != nil {
		return nil, fmt.Errorf("%s: %w", label, err)
	}
	d.window = newErrsRing(int(rateWindowSpan/sampleInterval) + 1)

	for i := 0; i < numStreams; i++ {
		et := uint16(etherBase + i)
		d.specs = append(d.specs, newFrameSpec(rx.mac, tx.mac, et, frameSizes))

		fd, err := openTxSocket(tx.idx)
		if err != nil {
			return nil, fmt.Errorf("%s tx socket: %w", label, err)
		}
		d.txFDs = append(d.txFDs, fd)

		fd, err = openRxSocket(rx.idx, et)
		if err != nil {
			return nil, fmt.Errorf("%s rx socket for 0x%04x: %w", label, et, err)
		}
		d.rxFDs = append(d.rxFDs, fd)
		d.rxStats = append(d.rxStats, &rxStats{})
	}

	return d, nil
}

func (d *direction) start(wg *sync.WaitGroup, done *atomic.Bool, rxReady *sync.WaitGroup, startTx <-chan struct{}) {
	for i, fd := range d.txFDs {
		w := &txWorker{
			fd:      fd,
			stream:  uint16(i),
			spec:    d.specs[i],
			batch:   batchSize,
			stats:   d.txStats[i],
			startTx: startTx,
		}
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer holdPanic()
			w.run(done)
		}()
	}
	for i, fd := range d.rxFDs {
		w := &rxWorker{
			fd:     fd,
			batch:  batchSize,
			stream: uint16(i),
			spec:   d.specs[i],
			stats:  d.rxStats[i],
			loss:   &d.streams[i],
			masked: d.masked,
			ready:  rxReady,
		}
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer holdPanic()
			w.run(done)
		}()
	}
}

func (d *direction) close() {
	for _, fd := range d.txFDs {
		unix.Close(fd)
	}
	for _, fd := range d.rxFDs {
		unix.Close(fd)
	}
	unix.Close(d.statFD)
}