Bugs-evident cleanup: cable-measure failure is fatal with its reason (failed/fail pretend-path and dead relinked plumbing deleted), bcm.command refuses partial SETs (0 or 5 params, stale-DATA trap made structural), BOOT gains a patched-ixgbe check naming stock-driver boots, panel fault rows matched by label and console rows strict on column count, phy.go split into sff/bcm/rollball and main.go's direction machinery into direction.go (pure moves), input events decoded via encoding/binary, gofmt; verified on hardware (20.00G, zero errors, ECD 41m)
This commit is contained in:
flamingcow
+420
@@ -0,0 +1,420 @@
|
||||
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
|
||||
|
||||
// While a cable measure runs, every failure counter in this direction is
|
||||
// suppressed at its source rather than counted, hidden and reverted.
|
||||
measuring *atomic.Bool
|
||||
|
||||
// Guards everything the sampler touches. The counters are read on their own
|
||||
// clock and drawn on another, and the two must not read them at once:
|
||||
// sampleDrops consumes what it reads, so a second caller would see a gap.
|
||||
mu sync.Mutex
|
||||
win *rateWindow
|
||||
drops uint64
|
||||
base counterSet
|
||||
|
||||
// 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
|
||||
|
||||
nic atomic.Uint64
|
||||
poller *nicPoller
|
||||
}
|
||||
|
||||
// Everything the display reads, taken at one instant, so a pair of these
|
||||
// describes both the rates and the errors over the span between them.
|
||||
type counterSet struct {
|
||||
t time.Time
|
||||
s sample
|
||||
drops uint64
|
||||
nic uint64
|
||||
}
|
||||
|
||||
func (d *direction) capture() counterSet {
|
||||
d.sampleDrops()
|
||||
s := d.snapshot()
|
||||
return counterSet{t: time.Now(), s: s, drops: d.drops, nic: d.nic.Load()}
|
||||
}
|
||||
|
||||
// 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,
|
||||
}
|
||||
}
|
||||
|
||||
// A ring of one bucket per drawn frame, spanning rateWindowSpan. Rates come
|
||||
// from the gap between adjacent buckets and errors from the ends of the ring,
|
||||
// so both slide forward every frame instead of stepping once a second.
|
||||
type rateWindow struct {
|
||||
buf []counterSet
|
||||
idx int
|
||||
filled bool
|
||||
}
|
||||
|
||||
func newRateWindow(n int) *rateWindow {
|
||||
return &rateWindow{buf: make([]counterSet, n)}
|
||||
}
|
||||
|
||||
func (w *rateWindow) push(c counterSet) {
|
||||
w.buf[w.idx] = c
|
||||
w.idx++
|
||||
if w.idx == len(w.buf) {
|
||||
w.idx = 0
|
||||
w.filled = true
|
||||
}
|
||||
}
|
||||
|
||||
func (w *rateWindow) count() int {
|
||||
if w.filled {
|
||||
return len(w.buf)
|
||||
}
|
||||
return w.idx
|
||||
}
|
||||
|
||||
// Indexed oldest first, so a partly filled ring reads the same as a full one.
|
||||
func (w *rateWindow) at(i int) counterSet {
|
||||
if w.filled {
|
||||
i += w.idx
|
||||
}
|
||||
return w.buf[i%len(w.buf)]
|
||||
}
|
||||
|
||||
// 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, so resetting just moves the origin
|
||||
// everything is measured from. Rates and the rolling error window are about now
|
||||
// rather than since the reset, so they keep running; the origin goes into the
|
||||
// ring so the newest bucket never sits behind it.
|
||||
func (d *direction) reset() {
|
||||
d.mu.Lock()
|
||||
d.base = d.capture()
|
||||
d.win.push(d.base)
|
||||
d.mu.Unlock()
|
||||
}
|
||||
|
||||
// The socket statistic is read-and-clear, so it is always consumed; a drop
|
||||
// during a cable measure is the blip's and is discarded at this source.
|
||||
func (d *direction) sampleDrops() {
|
||||
for _, fd := range d.rxFDs {
|
||||
n := packetDrops(fd)
|
||||
if !d.measuring.Load() {
|
||||
d.drops += n
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
func errsBetween(b, n counterSet) errs {
|
||||
return errs{
|
||||
lost: n.s.lost - b.s.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: (n.s.crcErr - b.s.crcErr) + (n.s.badHdr - b.s.badHdr) +
|
||||
(n.s.badMagic - b.s.badMagic) + (n.s.badLen - b.s.badLen),
|
||||
// What the hardware reported. Nothing the host declined to send is here,
|
||||
// so this one going red means the cable.
|
||||
link: (n.nic - b.nic) + (n.s.rxErrs - b.s.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: (n.drops - b.drops) + (n.s.late - b.s.late) +
|
||||
(n.s.txErrs - b.s.txErrs),
|
||||
}
|
||||
}
|
||||
|
||||
func (d *direction) counters(now counterSet) view {
|
||||
return view{
|
||||
rxFrames: now.s.rxFrames - d.base.s.rxFrames,
|
||||
rxBytes: now.s.rxBytes - d.base.s.rxBytes,
|
||||
since: errsBetween(d.base, now),
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
func (d *direction) sample() {
|
||||
d.mu.Lock()
|
||||
d.win.push(d.capture())
|
||||
d.readRateBucket(time.Now())
|
||||
d.mu.Unlock()
|
||||
}
|
||||
|
||||
// Draws what the sampler last put in the ring rather than reading the counters
|
||||
// again, so the display never participates in the measurement.
|
||||
func (d *direction) displayView() view {
|
||||
d.mu.Lock()
|
||||
n := d.win.count()
|
||||
if n == 0 {
|
||||
d.mu.Unlock()
|
||||
return view{}
|
||||
}
|
||||
v := d.counters(d.win.at(n - 1))
|
||||
if n >= 2 {
|
||||
v.window = errsBetween(d.win.at(0), d.win.at(n-1))
|
||||
}
|
||||
v.rxPPS = float64(d.rateFrames) / rateBucketSecs
|
||||
v.rxGbps = gbps(d.rateBytes, d.rateFrames, rateBucketSecs)
|
||||
d.mu.Unlock()
|
||||
return v
|
||||
}
|
||||
|
||||
// Whatever the interfaces counted before now is not ours, and no interval has
|
||||
// elapsed yet, so every baseline starts here and nothing is reported until the
|
||||
// first one completes.
|
||||
func (d *direction) primeCounters() {
|
||||
d.poller.prime()
|
||||
d.reset()
|
||||
}
|
||||
|
||||
func buildDirection(label string, tx, rx endpoint, measuring *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),
|
||||
measuring: measuring,
|
||||
}
|
||||
// Held open for the life of the run: the stats ioctl is issued five times a
|
||||
// second 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.poller, err = newNICPoller(statFD, tx.name, rx.name, &d.nic, measuring)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", label, err)
|
||||
}
|
||||
d.win = newRateWindow(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],
|
||||
measuring: d.measuring,
|
||||
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],
|
||||
measuring: d.measuring,
|
||||
ready: rxReady,
|
||||
}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer holdPanic()
|
||||
w.run(done)
|
||||
}()
|
||||
}
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer holdPanic()
|
||||
d.poller.run(done, startTx)
|
||||
}()
|
||||
}
|
||||
|
||||
func (d *direction) close() {
|
||||
for _, fd := range d.txFDs {
|
||||
unix.Close(fd)
|
||||
}
|
||||
for _, fd := range d.rxFDs {
|
||||
unix.Close(fd)
|
||||
}
|
||||
unix.Close(d.statFD)
|
||||
}
|
||||
Reference in New Issue
Block a user