Compare commits

..
3 Commits
5 changed files with 56 additions and 28 deletions
+19 -2
View File
@@ -5,6 +5,7 @@ import (
"fmt" "fmt"
"math" "math"
"path/filepath" "path/filepath"
"sync/atomic"
"unsafe" "unsafe"
"golang.org/x/sys/unix" "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 // 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. // a timer, so drawing is paced by the panel instead of by a guess at its rate.
flips chan struct{} 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 { 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. // 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() { func (fb *framebuffer) readEvents() {
buf := make([]byte, 4096) buf := make([]byte, 4096)
for { for {
n, err := unix.Read(fb.fd, buf) n, err := unix.Read(fb.fd, buf)
if n <= 0 || err != nil { if fb.closing.Load() {
return 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; { for off := 0; off+8 <= n; {
typ := binary.LittleEndian.Uint32(buf[off:]) typ := binary.LittleEndian.Uint32(buf[off:])
length := int(binary.LittleEndian.Uint32(buf[off+4:])) length := int(binary.LittleEndian.Uint32(buf[off+4:]))
if length < 8 || off+length > n { 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 { if typ == drm.EventFlipComplete {
select { select {
@@ -300,6 +316,7 @@ func (fb *framebuffer) readEvents() {
} }
func (fb *framebuffer) close() { func (fb *framebuffer) close() {
fb.closing.Store(true)
for i := range fb.bufs { for i := range fb.bufs {
if fb.bufs[i].mem != nil { if fb.bufs[i].mem != nil {
unix.Munmap(fb.bufs[i].mem) unix.Munmap(fb.bufs[i].mem)
+5 -4
View File
@@ -36,8 +36,8 @@ func newLossWindows(tx []*txStats) []lossWindow {
return w return w
} }
// A sequence number is only judged once it falls out of the window, so frames // A sequence number is only judged once it falls out of the window, so a frame
// still queued in another rx worker are never miscounted as lost. // 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 // 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 // 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() w.mu.Lock()
if !w.inited { if !w.inited {
// Start half a window below the first sequence seen, so frames another // Start half a window below the first sequence seen, so anything the
// rx worker is still holding land inside the window rather than late. // sender put on the wire before it lands inside the window rather than
// below the base.
if seq > lossSlots/2 { if seq > lossSlots/2 {
w.base = seq - lossSlots/2 w.base = seq - lossSlots/2
} }
+16 -16
View File
@@ -437,7 +437,7 @@ func buildDirection(label string, tx, rx endpoint) (*direction, error) {
return d, nil 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 { for i, fd := range d.txFDs {
w := &txWorker{ w := &txWorker{
fd: fd, fd: fd,
@@ -450,7 +450,7 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, rxRea
wg.Add(1) wg.Add(1)
go func() { go func() {
defer wg.Done() defer wg.Done()
w.run(doneTx) w.run(done)
}() }()
} }
for i, fd := range d.rxFDs { 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) wg.Add(1)
go func() { go func() {
defer wg.Done() 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) wg.Add(1)
go func() { go func() {
defer wg.Done() defer wg.Done()
sender.run(doneTx, startTx) sender.run(done, startTx)
}() }()
receiver := &probeReceiver{fd: d.probeRxFD, stats: d.cable, ready: rxReady} receiver := &probeReceiver{fd: d.probeRxFD, stats: d.cable, ready: rxReady}
wg.Add(1) wg.Add(1)
go func() { go func() {
defer wg.Done() defer wg.Done()
receiver.run(doneRx) receiver.run(done)
}() }()
wg.Add(1) wg.Add(1)
go func() { go func() {
defer wg.Done() defer wg.Done()
d.poller.run(doneRx, startTx) d.poller.run(done, startTx)
}() }()
} }
@@ -514,7 +514,9 @@ const (
linkSpeed = 10.0 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() { func main() {
// The names the kernel gives the only two ports built into it, since as // 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 var linkRows [][]string
for _, e := range []endpoint{a, b} { for _, e := range []endpoint{a, b} {
linkRows = append(linkRows, []string{ linkRows = append(linkRows, []string{
paint(e.tag, cCyan), e.name, e.macString(), paint(e.tag, cCyan), e.name, e.macString(), fmt.Sprintf("%d", e.mtu),
fmt.Sprintf("%.0f Gb/s", linkSpeed), fmt.Sprintf("%d", e.mtu),
}) })
} }
fmt.Println(renderBox("LINKS", fmt.Println(renderBox("LINKS",
[]string{"TAG", "INTERFACE", "MAC", "SPEED", "MTU"}, []string{"TAG", "INTERFACE", "MAC", "MTU"},
[]bool{false, false, false, true, true}, linkRows)) []bool{false, false, false, true}, linkRows))
fmt.Println() fmt.Println()
// One row carries both directions, so line rate is both links at once. // One row carries both directions, so line rate is both links at once.
target := linkSpeed * float64(len(dirs)) target := linkSpeed * float64(len(dirs))
var doneTx, doneRx atomic.Bool var done atomic.Bool
var wg sync.WaitGroup var wg sync.WaitGroup
var rxReady sync.WaitGroup var rxReady sync.WaitGroup
startTx := make(chan struct{}) startTx := make(chan struct{})
@@ -631,13 +632,13 @@ func run(aName, bName string, nsPerM float64) error {
rxReady.Add(len(d.rxFDs) + 1) rxReady.Add(len(d.rxFDs) + 1)
} }
for _, d := range dirs { for _, d := range dirs {
d.start(&wg, &doneTx, &doneRx, &rxReady, startTx) d.start(&wg, &done, &rxReady, startTx)
} }
samp := &sampler{dirs: dirs} samp := &sampler{dirs: dirs}
wg.Add(1) wg.Add(1)
go func() { go func() {
defer wg.Done() defer wg.Done()
samp.run(&doneRx, startTx) samp.run(&done, startTx)
}() }()
rxReady.Wait() rxReady.Wait()
@@ -673,8 +674,7 @@ func run(aName, bName string, nsPerM float64) error {
for { for {
select { select {
case <-sig: case <-sig:
doneTx.Store(true) done.Store(true)
doneRx.Store(true)
wg.Wait() wg.Wait()
return nil return nil
case <-space: case <-space:
+2 -3
View File
@@ -15,9 +15,8 @@ type mmsghdr struct {
func htons(v uint16) uint16 { return v<<8 | v>>8 } func htons(v uint16) uint16 { return v<<8 | v>>8 }
// Set through the FORCE options alone: the plain ones are clamped to wmem_max // Set through the FORCE options alone, since the plain ones are clamped to
// and rmem_max, so falling back to them would quietly leave a fraction of this // wmem_max and rmem_max and would quietly leave a fraction of this.
// and go on measuring as though it had not.
const ( const (
sndbufBytes = 8 << 20 sndbufBytes = 8 << 20
rcvbufBytes = 64 << 20 rcvbufBytes = 64 << 20
+14 -3
View File
@@ -83,6 +83,11 @@ func allRuleLocations(fd int, ifname string) ([]uint32, uint32, error) {
return nil, 0, err 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:] raw := buf[locOff:]
locs := make([]uint32, 0, all.ruleCnt) locs := make([]uint32, 0, all.ruleCnt)
for i := 0; i < int(all.ruleCnt); i++ { for i := 0; i < int(all.ruleCnt); i++ {
@@ -171,9 +176,16 @@ func checkFlowRules(fd int, ifname string, ethertypes []uint16) checkResult {
taken[loc] = true 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 next := capacity - 1
for i, et := range ethertypes { 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-- next--
} }
if err := insertEtherRule(fd, ifname, et, uint64(i), next); err != nil { 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 // 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 // under test, so with no cable there is never going to be one.
// result to report rather than a reason to refuse to start.
func configureSystem(ifnames []string, ethertypes []uint16) []checkResult { func configureSystem(ifnames []string, ethertypes []uint16) []checkResult {
return withIoctlSocket(func(fd int) []checkResult { return withIoctlSocket(func(fd int) []checkResult {
out := []checkResult{checkGovernor(wantGovernor)} out := []checkResult{checkGovernor(wantGovernor)}