diff --git a/frame.go b/frame.go index 4a8fef5..1980bf1 100644 --- a/frame.go +++ b/frame.go @@ -13,6 +13,8 @@ const ( hdrMagic = 0x43424c54 minFrame = ethHdrLen + hdrLen maxFrame = 9216 + + hdrCheckOff = 18 ) var crcTable = crc32.MakeTable(crc32.Castagnoli) @@ -113,6 +115,26 @@ func (f *frameSpec) expectedCRC(patIdx, payLen int) (uint32, bool) { return crc, ok } +// A one's complement sum over the header, skipping the two bytes it is written +// into. The payload checksum covers the payload alone, so without this the +// sequence number is the one field on the wire that nothing checks, and a bit +// flipped in it arrives looking exactly like a legitimate frame from far in the +// future. Every single bit error changes the folded sum, which is the error a +// cable going marginal actually produces. +func headerCheck(h []byte) uint16 { + var sum uint32 + for i := 0; i+1 < hdrLen; i += 2 { + if i == hdrCheckOff { + continue + } + sum += uint32(binary.BigEndian.Uint16(h[i:])) + } + for sum>>16 != 0 { + sum = sum&0xffff + sum>>16 + } + return ^uint16(sum) +} + func putHeader(buf []byte, patIdx int, stream uint16, seq uint64, payLen int) { h := buf[ethHdrLen:] binary.BigEndian.PutUint32(h[0:4], hdrMagic) @@ -121,7 +143,7 @@ func putHeader(buf []byte, patIdx int, stream uint16, seq uint64, payLen int) { binary.BigEndian.PutUint16(h[6:8], stream) binary.BigEndian.PutUint64(h[8:16], seq) binary.BigEndian.PutUint16(h[16:18], uint16(payLen)) - binary.BigEndian.PutUint16(h[18:20], 0) + binary.BigEndian.PutUint16(h[hdrCheckOff:], headerCheck(h)) } type parsed struct { @@ -131,21 +153,35 @@ type parsed struct { payLen int } -func parseHeader(buf []byte) (parsed, bool) { +// Why a frame was turned away. A header that is not ours and a header of ours +// that arrived damaged are different things to report: the first is somebody +// else on the wire, the second is the fault being tested for. +type hdrStatus int + +const ( + hdrOK hdrStatus = iota + hdrForeign + hdrDamaged +) + +func parseHeader(buf []byte) (parsed, hdrStatus) { var p parsed if len(buf) < minFrame { - return p, false + return p, hdrForeign } h := buf[ethHdrLen:] if binary.BigEndian.Uint32(h[0:4]) != hdrMagic { - return p, false + return p, hdrForeign + } + if binary.BigEndian.Uint16(h[hdrCheckOff:]) != headerCheck(h) { + return p, hdrDamaged } p.patIdx = int(h[5]) p.stream = binary.BigEndian.Uint16(h[6:8]) p.seq = binary.BigEndian.Uint64(h[8:16]) p.payLen = int(binary.BigEndian.Uint16(h[16:18])) if minFrame+p.payLen > len(buf) { - return p, false + return p, hdrDamaged } - return p, true + return p, hdrOK } diff --git a/frame_test.go b/frame_test.go new file mode 100644 index 0000000..1a52ec4 --- /dev/null +++ b/frame_test.go @@ -0,0 +1,66 @@ +package main + +import "testing" + +func testFrame() []byte { + buf := make([]byte, 512) + spec := newFrameSpec([6]byte{1, 2, 3, 4, 5, 6}, [6]byte{7, 8, 9, 10, 11, 12}, + etherBase, []int{512}) + spec.prefill(buf, 3) + putHeader(buf, 3, 5, 0x0123456789ab, 512-minFrame) + return buf +} + +func TestHeaderRoundTrip(t *testing.T) { + p, st := parseHeader(testFrame()) + if st != hdrOK { + t.Fatalf("status = %v, want hdrOK", st) + } + want := parsed{patIdx: 3, stream: 5, seq: 0x0123456789ab, payLen: 512 - minFrame} + if p != want { + t.Errorf("parsed = %+v, want %+v", p, want) + } +} + +// The sequence number is the one field on the wire that nothing else checks, and +// a single flipped bit in it is exactly what a marginal cable produces. Every +// bit of the header has to be covered, including the checksum's own. +func TestHeaderChecksumCatchesEverySingleBitFlip(t *testing.T) { + for byteIdx := 0; byteIdx < hdrLen; byteIdx++ { + for bit := 0; bit < 8; bit++ { + buf := testFrame() + buf[ethHdrLen+byteIdx] ^= 1 << bit + + _, st := parseHeader(buf) + if st == hdrOK { + t.Errorf("header byte %d bit %d flipped and the frame was accepted", + byteIdx, bit) + } + // The magic is what says the frame is ours at all, so damage there is + // somebody else's traffic as far as we can tell. + wantForeign := byteIdx < 4 + if wantForeign && st != hdrForeign { + t.Errorf("magic byte %d bit %d = %v, want hdrForeign", byteIdx, bit, st) + } + if !wantForeign && st != hdrDamaged { + t.Errorf("header byte %d bit %d = %v, want hdrDamaged", byteIdx, bit, st) + } + } + } +} + +func TestHeaderShortFrameIsForeign(t *testing.T) { + if _, st := parseHeader(testFrame()[:minFrame-1]); st != hdrForeign { + t.Errorf("status = %v, want hdrForeign", st) + } +} + +// A length that runs past the frame is damage rather than a foreign header: the +// magic and the checksum both said the header was ours. +func TestHeaderLengthPastFrameIsDamaged(t *testing.T) { + buf := testFrame() + putHeader(buf, 3, 5, 1, len(buf)) + if _, st := parseHeader(buf); st != hdrDamaged { + t.Errorf("status = %v, want hdrDamaged", st) + } +} diff --git a/loss.go b/loss.go index e22f2bb..ed48417 100644 --- a/loss.go +++ b/loss.go @@ -18,19 +18,39 @@ type lossWindow struct { bits []uint64 lost atomic.Uint64 late atomic.Uint64 + + // The sending half of this stream. Both halves are in this process and the + // receiving socket ignores its own outgoing frames, so this window is fed by + // that sender alone and by nothing else on the wire. + sent *atomic.Uint64 } -func newLossWindows(n int) []lossWindow { - w := make([]lossWindow, n) +// Takes the sending halves rather than a count, so a window cannot be built +// without the bound it judges sequence numbers against. +func newLossWindows(tx []*txStats) []lossWindow { + w := make([]lossWindow, len(tx)) for i := range w { w[i].bits = make([]uint64, lossWords) + w[i].sent = &tx[i].sent } 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. -func (w *lossWindow) observe(seq uint64) { +// +// 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 +// is refused before it can touch the base. Believing one would drag the window +// up past every real sequence, write off the span in between as lost, and leave +// every frame after it arriving below the base and counted late for the rest of +// the run. Nothing that arrives later is evidence enough to undo that, which is +// why the sender's own frontier is the guard rather than a limit on how far a +// single step may move. +func (w *lossWindow) observe(seq uint64) bool { + if seq >= w.sent.Load() { + return false + } w.mu.Lock() if !w.inited { // Start half a window below the first sequence seen, so frames another @@ -43,7 +63,7 @@ func (w *lossWindow) observe(seq uint64) { if seq < w.base { w.mu.Unlock() w.late.Add(1) - return + return true } if seq >= w.base+lossSlots { w.evict(seq - lossSlots + 1) @@ -51,6 +71,7 @@ func (w *lossWindow) observe(seq uint64) { idx := seq & (lossSlots - 1) w.bits[idx>>6] |= 1 << (idx & 63) w.mu.Unlock() + return true } func (w *lossWindow) evict(newBase uint64) { diff --git a/loss_test.go b/loss_test.go index b88616b..5f8bd3e 100644 --- a/loss_test.go +++ b/loss_test.go @@ -2,13 +2,17 @@ package main import "testing" -func newWindow() *lossWindow { - w := newLossWindows(1) +// sent is the sender's frontier: sequence numbers at or above it were never put +// on the wire. +func newWindow(sent uint64) *lossWindow { + tx := &txStats{} + tx.sent.Store(sent) + w := newLossWindows([]*txStats{tx}) return &w[0] } func TestLossWindowContiguousLosesNothing(t *testing.T) { - w := newWindow() + w := newWindow(lossSlots + 1000) for seq := uint64(0); seq < lossSlots+1000; seq++ { w.observe(seq) } @@ -21,7 +25,7 @@ func TestLossWindowContiguousLosesNothing(t *testing.T) { } func TestLossWindowCountsGapOnceEvicted(t *testing.T) { - w := newWindow() + w := newWindow(lossSlots + 1000) for seq := uint64(0); seq < lossSlots+1000; seq++ { if seq == 100 { continue @@ -36,7 +40,7 @@ func TestLossWindowCountsGapOnceEvicted(t *testing.T) { // Arriving out of order inside the window is not loss: a sequence is only // judged once it falls out the far end. func TestLossWindowOutOfOrderIsNotLoss(t *testing.T) { - w := newWindow() + w := newWindow(lossSlots + 1000) for seq := uint64(99); ; seq-- { w.observe(seq) if seq == 0 { @@ -54,7 +58,7 @@ func TestLossWindowOutOfOrderIsNotLoss(t *testing.T) { // The other branch of evict: a jump past a whole window writes off everything // the window held plus the sequences that never landed in it at all. func TestLossWindowJumpBeyondWindow(t *testing.T) { - w := newWindow() + w := newWindow(200001) w.observe(0) w.observe(200000) @@ -69,7 +73,7 @@ func TestLossWindowJumpBeyondWindow(t *testing.T) { } func TestLossWindowBelowBaseIsLate(t *testing.T) { - w := newWindow() + w := newWindow(100001) w.observe(100000) w.observe(1000) if got := w.late.Load(); got != 1 { @@ -83,9 +87,58 @@ func TestLossWindowBelowBaseIsLate(t *testing.T) { // The first sequence seen starts the window half a span below it, so frames // another worker is still holding land inside rather than arriving late. func TestLossWindowStartsHalfAWindowBack(t *testing.T) { - w := newWindow() + w := newWindow(100001) w.observe(100000) if w.base != 100000-lossSlots/2 { t.Errorf("base = %d, want %d", w.base, 100000-lossSlots/2) } } + +// The failure this guard exists for. A sequence number the sender never reached +// used to drag the base up to it, write off the span in between as lost, and +// leave every real frame after it below the base and counted late for the rest +// of the run. It has to be refused outright, and the stream has to go on +// counting as if it had never arrived. +func TestLossWindowRefusesUnsentSeq(t *testing.T) { + const first = 100000 + w := newWindow(first + 2000) + for seq := uint64(first); seq < first+1000; seq++ { + w.observe(seq) + } + base := w.base + + if w.observe(1 << 62) { + t.Error("a sequence number far past the sender's frontier was accepted") + } + if got := w.lost.Load(); got != 0 { + t.Errorf("lost = %d after one impossible sequence number, want 0", got) + } + if w.base != base { + t.Errorf("base moved to %d, want it left at %d where the real traffic put it", + w.base, base) + } + + // Still tracking the real traffic, rather than reporting every frame late + // against a base that ran away. + for seq := uint64(first + 1000); seq < first+2000; seq++ { + w.observe(seq) + } + if got := w.late.Load(); got != 0 { + t.Errorf("late = %d, want 0", got) + } + if got := w.lost.Load(); got != 0 { + t.Errorf("lost = %d, want 0", got) + } +} + +// The sender's own frontier is the bound, so the sequence one past it is +// refused while the one below it is not. +func TestLossWindowBoundIsExclusive(t *testing.T) { + w := newWindow(500) + if !w.observe(499) { + t.Error("the last sequence the sender put on the wire was refused") + } + if w.observe(500) { + t.Error("a sequence the sender had not reached was accepted") + } +} diff --git a/main.go b/main.go index 290c922..a0e0daf 100644 --- a/main.go +++ b/main.go @@ -169,6 +169,7 @@ type sample struct { rxFrames, rxBytes uint64 lost, late uint64 crcErr, badMagic uint64 + badHdr uint64 badLen uint64 txErrs uint64 rxErrs uint64 @@ -204,6 +205,7 @@ func (d *direction) snapshot() sample { 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() } @@ -279,10 +281,11 @@ type view struct { func errsBetween(b, n counterSet) errs { return errs{ lost: n.s.lost - b.s.lost, - // Three ways of noticing one thing: a payload that does not match its - // checksum, a header that is not ours, and a length that cannot be. - corrupt: (n.s.crcErr - b.s.crcErr) + (n.s.badMagic - b.s.badMagic) + - (n.s.badLen - b.s.badLen), + // 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), @@ -372,8 +375,15 @@ func (d *direction) primeCounters() { } func buildDirection(label string, tx, rx endpoint) (*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{ - streams: newLossWindows(numStreams), + txStats: txs, + streams: newLossWindows(txs), cable: newCableStats(), } // Held open for the life of the run: the stats ioctl is issued five times a @@ -399,7 +409,6 @@ func buildDirection(label string, tx, rx endpoint) (*direction, error) { return nil, fmt.Errorf("%s tx socket: %w", label, err) } d.txFDs = append(d.txFDs, fd) - d.txStats = append(d.txStats, &txStats{}) fd, err = openRxSocket(rx.idx, et) if err != nil { diff --git a/main_test.go b/main_test.go index 657e379..2afa603 100644 --- a/main_test.go +++ b/main_test.go @@ -95,19 +95,19 @@ func TestErrsBetweenBuckets(t *testing.T) { n := counterSet{ s: sample{ lost: 1, late: 7, - crcErr: 2, badMagic: 3, badLen: 4, + crcErr: 2, badMagic: 3, badLen: 4, badHdr: 10, txErrs: 6, rxErrs: 5, }, drops: 9, nic: 8, } got := errsBetween(counterSet{}, n) - want := errs{lost: 1, corrupt: 2 + 3 + 4, link: 8 + 5, internal: 9 + 7 + 6} + want := errs{lost: 1, corrupt: 2 + 3 + 4 + 10, link: 8 + 5, internal: 9 + 7 + 6} if got != want { t.Errorf("errsBetween = %+v, want %+v", got, want) } - if got.total() != 45 { - t.Errorf("total = %d, want 45", got.total()) + if got.total() != 55 { + t.Errorf("total = %d, want 55", got.total()) } } diff --git a/probe.go b/probe.go index 308f9c8..b84837e 100644 --- a/probe.go +++ b/probe.go @@ -210,8 +210,8 @@ func (r *probeReceiver) run(done *atomic.Bool) { if err != nil { continue } - h, ok := parseHeader(buf[:n]) - if !ok || h.stream != probeStream { + h, st := parseHeader(buf[:n]) + if st != hdrOK || h.stream != probeStream { continue } ts, ok := hwTimestamp(oob[:oobn]) diff --git a/rx.go b/rx.go index 3149d59..902d876 100644 --- a/rx.go +++ b/rx.go @@ -12,6 +12,7 @@ type rxStats struct { frames atomic.Uint64 bytes atomic.Uint64 badMagic atomic.Uint64 + badHdr atomic.Uint64 badLen atomic.Uint64 crcErr atomic.Uint64 rxErrs atomic.Uint64 @@ -99,9 +100,13 @@ func (w *rxWorker) run(done *atomic.Bool) { } for i := 0; i < n; i++ { buf := bufs[i][:int(hdrs[i].len)] - p, ok := parseHeader(buf) - if !ok { - w.stats.badMagic.Add(1) + p, st := parseHeader(buf) + if st != hdrOK { + if st == hdrForeign { + w.stats.badMagic.Add(1) + } else { + w.stats.badHdr.Add(1) + } continue } w.stats.frames.Add(1) @@ -110,8 +115,13 @@ func (w *rxWorker) run(done *atomic.Bool) { w.stats.observe(ts, uint64(len(buf))) } - if int(p.stream) < len(w.streams) { - w.streams[p.stream].observe(p.seq) + // A sequence number the sender never reached got past the header + // checksum, so the frame is damaged whatever its payload says. Counted + // here rather than left to the payload check, which would report the + // same frame twice or, if only the header was hit, not at all. + if int(p.stream) < len(w.streams) && !w.streams[p.stream].observe(p.seq) { + w.stats.badHdr.Add(1) + continue } want, ok := w.spec.expectedCRC(p.patIdx, p.payLen) diff --git a/tx.go b/tx.go index a96f2c4..59adc8f 100644 --- a/tx.go +++ b/tx.go @@ -8,6 +8,12 @@ import ( type txStats struct { errs atomic.Uint64 + + // The exclusive upper bound on any sequence number this stream can have put + // on the wire. Read by the receiving half of the same stream, which is the + // only thing that can tell a gap it must write off from a sequence number + // that was never sent at all. + sent atomic.Uint64 } type txWorker struct { @@ -46,6 +52,11 @@ func (w *txWorker) run(done *atomic.Bool) { iovs[i].Len = uint64(size) } + // Published before the send rather than after it, so it is never behind a + // frame already in flight. Overshooting a partial send is harmless: it is + // a bound, and one the next pass raises again. + w.stats.sent.Store(seq + uint64(w.batch)) + n, err := sendmmsg(w.fd, hdrs) if n > 0 { seq += uint64(n)