Fold the shutdown flags into one, take the smallest frame to the wire minimum, drop the speed column

fb.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"math"
 	"path/filepath"
+	"sync/atomic"
 	"unsafe"
 
 	"golang.org/x/sys/unix"
@@ -49,6 +50,9 @@ type framebuffer struct {
 	// One token per completed flip. The render loop waits on this rather than on
 	// a timer, so drawing is paced by the panel instead of by a guess at its rate.
 	flips chan struct{}
+
+	// Set before the fd goes, so the event reader can tell shutdown from failure.
+	closing atomic.Bool
 }
 
 func (fb *framebuffer) offset(x, y int) int {
@@ -275,18 +279,30 @@ func openFramebuffer() (*framebuffer, error) {
 }
 
 // Flip completions arrive on the drm fd as a stream of length-prefixed events.
+// Nothing else feeds fb.flips, so returning early here freezes the panel on its
+// last frame while everything else goes on running.
 func (fb *framebuffer) readEvents() {
 	buf := make([]byte, 4096)
 	for {
 		n, err := unix.Read(fb.fd, buf)
-		if n <= 0 || err != nil {
+		if fb.closing.Load() {
 			return
 		}
+		if err == unix.EINTR || err == unix.EAGAIN {
+			continue
+		}
+		if err != nil {
+			panic(fmt.Sprintf("reading drm events: %v", err))
+		}
+		if n == 0 {
+			panic("drm fd reported end of file")
+		}
 		for off := 0; off+8 <= n; {
 			typ := binary.LittleEndian.Uint32(buf[off:])
 			length := int(binary.LittleEndian.Uint32(buf[off+4:]))
 			if length < 8 || off+length > n {
-				return
+				panic(fmt.Sprintf("drm event at offset %d claims %d bytes of %d read",
+					off, length, n))
 			}
 			if typ == drm.EventFlipComplete {
 				select {
@@ -300,6 +316,7 @@ func (fb *framebuffer) readEvents() {
 }
 
 func (fb *framebuffer) close() {
+	fb.closing.Store(true)
 	for i := range fb.bufs {
 		if fb.bufs[i].mem != nil {
 			unix.Munmap(fb.bufs[i].mem)

loss.go

@@ -36,8 +36,8 @@ func newLossWindows(tx []*txStats) []lossWindow {
 	return w
 }
 
-// A sequence number is only judged once it falls out of the window, so frames
-// still queued in another rx worker are never miscounted as lost.
+// A sequence number is only judged once it falls out of the window, so a frame
+// that has arrived but not yet been drained is never miscounted as lost.
 //
 // Reports whether the sequence number could have come off the wire at all. One
 // above what the sender has reached is a damaged header rather than a gap, and
@@ -53,8 +53,9 @@ func (w *lossWindow) observe(seq uint64) bool {
 	}
 	w.mu.Lock()
 	if !w.inited {
-		// Start half a window below the first sequence seen, so frames another
-		// rx worker is still holding land inside the window rather than late.
+		// Start half a window below the first sequence seen, so anything the
+		// sender put on the wire before it lands inside the window rather than
+		// below the base.
 		if seq > lossSlots/2 {
 			w.base = seq - lossSlots/2
 		}

main.go

@@ -437,7 +437,7 @@ func buildDirection(label string, tx, rx endpoint) (*direction, error) {
 	return d, nil
 }
 
-func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, rxReady *sync.WaitGroup, startTx <-chan struct{}) {
+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,
@@ -450,7 +450,7 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, rxRea
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			w.run(doneTx)
+			w.run(done)
 		}()
 	}
 	for i, fd := range d.rxFDs {
@@ -466,7 +466,7 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, rxRea
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			w.run(doneRx)
+			w.run(done)
 		}()
 	}
 
@@ -474,20 +474,20 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, rxRea
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		sender.run(doneTx, startTx)
+		sender.run(done, startTx)
 	}()
 
 	receiver := &probeReceiver{fd: d.probeRxFD, stats: d.cable, ready: rxReady}
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		receiver.run(doneRx)
+		receiver.run(done)
 	}()
 
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		d.poller.run(doneRx, startTx)
+		d.poller.run(done, startTx)
 	}()
 }
 
@@ -514,7 +514,9 @@ const (
 	linkSpeed = 10.0
 )
 
-var frameSizes = []int{64, 128, 256, 512, 1024, 1280, 1514}
+// The mac appends the fcs, so 60 and 1514 here are the smallest and largest
+// standard frames, 64 and 1518 on the wire.
+var frameSizes = []int{60, 128, 256, 512, 1024, 1280, 1514}
 
 func main() {
 	// The names the kernel gives the only two ports built into it, since as
@@ -611,19 +613,18 @@ func run(aName, bName string, nsPerM float64) error {
 	var linkRows [][]string
 	for _, e := range []endpoint{a, b} {
 		linkRows = append(linkRows, []string{
-			paint(e.tag, cCyan), e.name, e.macString(),
-			fmt.Sprintf("%.0f Gb/s", linkSpeed), fmt.Sprintf("%d", e.mtu),
+			paint(e.tag, cCyan), e.name, e.macString(), fmt.Sprintf("%d", e.mtu),
 		})
 	}
 	fmt.Println(renderBox("LINKS",
-		[]string{"TAG", "INTERFACE", "MAC", "SPEED", "MTU"},
-		[]bool{false, false, false, true, true}, linkRows))
+		[]string{"TAG", "INTERFACE", "MAC", "MTU"},
+		[]bool{false, false, false, true}, linkRows))
 	fmt.Println()
 
 	// One row carries both directions, so line rate is both links at once.
 	target := linkSpeed * float64(len(dirs))
 
-	var doneTx, doneRx atomic.Bool
+	var done atomic.Bool
 	var wg sync.WaitGroup
 	var rxReady sync.WaitGroup
 	startTx := make(chan struct{})
@@ -631,13 +632,13 @@ func run(aName, bName string, nsPerM float64) error {
 		rxReady.Add(len(d.rxFDs) + 1)
 	}
 	for _, d := range dirs {
-		d.start(&wg, &doneTx, &doneRx, &rxReady, startTx)
+		d.start(&wg, &done, &rxReady, startTx)
 	}
 	samp := &sampler{dirs: dirs}
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		samp.run(&doneRx, startTx)
+		samp.run(&done, startTx)
 	}()
 	rxReady.Wait()
 
@@ -673,8 +674,7 @@ func run(aName, bName string, nsPerM float64) error {
 	for {
 		select {
 		case <-sig:
-			doneTx.Store(true)
-			doneRx.Store(true)
+			done.Store(true)
 			wg.Wait()
 			return nil
 		case <-space:

sock.go

@@ -15,9 +15,8 @@ type mmsghdr struct {
 
 func htons(v uint16) uint16 { return v<<8 | v>>8 }
 
-// Set through the FORCE options alone: the plain ones are clamped to wmem_max
-// and rmem_max, so falling back to them would quietly leave a fraction of this
-// and go on measuring as though it had not.
+// Set through the FORCE options alone, since the plain ones are clamped to
+// wmem_max and rmem_max and would quietly leave a fraction of this.
 const (
 	sndbufBytes = 8 << 20
 	rcvbufBytes = 64 << 20

system.go

@@ -83,6 +83,11 @@ func allRuleLocations(fd int, ifname string) ([]uint32, uint32, error) {
 		return nil, 0, err
 	}
 
+	if all.ruleCnt > cnt.ruleCnt {
+		return nil, 0, fmt.Errorf("%s reported %d filter locations into room for %d",
+			ifname, all.ruleCnt, cnt.ruleCnt)
+	}
+
 	raw := buf[locOff:]
 	locs := make([]uint32, 0, all.ruleCnt)
 	for i := 0; i < int(all.ruleCnt); i++ {
@@ -171,9 +176,16 @@ func checkFlowRules(fd int, ifname string, ethertypes []uint16) checkResult {
 		taken[loc] = true
 	}
 
+	// Inserting at a taken location would evict it, and would then leave every
+	// later ethertype evicting the one before it at that same location.
 	next := capacity - 1
 	for i, et := range ethertypes {
-		for taken[next] && next > 0 {
+		for taken[next] {
+			if next == 0 {
+				res.err = fmt.Errorf("no free filter location below %d for ethertype 0x%04x",
+					capacity, et)
+				return res
+			}
 			next--
 		}
 		if err := insertEtherRule(fd, ifname, et, uint64(i), next); err != nil {
@@ -586,8 +598,7 @@ func withIoctlSocket(fn func(fd int) []checkResult) []checkResult {
 }
 
 // Nothing here waits for a carrier: the two ports are the two ends of the cable
-// under test, so with no cable there will never be one, and a dead wire is a
-// result to report rather than a reason to refuse to start.
+// under test, so with no cable there is never going to be one.
 func configureSystem(ifnames []string, ethertypes []uint16) []checkResult {
 	return withIoctlSocket(func(fd int) []checkResult {
 		out := []checkResult{checkGovernor(wantGovernor)}