Compare commits

...
2 Commits
5 changed files with 32 additions and 33 deletions
+13 -6
View File
@@ -9,7 +9,7 @@ import (
const ( const (
etherBase = 0x88b5 etherBase = 0x88b5
ethHdrLen = 14 ethHdrLen = 14
hdrLen = 24 hdrLen = 20
hdrMagic = 0x43424c54 hdrMagic = 0x43424c54
minFrame = ethHdrLen + hdrLen minFrame = ethHdrLen + hdrLen
maxFrame = 9216 maxFrame = 9216
@@ -88,7 +88,7 @@ func newFrameSpec(dst, src [6]byte, etherType uint16, sizes []int) *frameSpec {
fill(ref) fill(ref)
crcFor := make(map[int]uint32, len(sizes)) crcFor := make(map[int]uint32, len(sizes))
for _, s := range sizes { for _, s := range sizes {
crcFor[s] = crc32.Checksum(ref[:s-minFrame], crcTable) crcFor[s-minFrame] = crc32.Checksum(ref[:s-minFrame], crcTable)
} }
f.refs = append(f.refs, ref) f.refs = append(f.refs, ref)
f.crcFor = append(f.crcFor, crcFor) f.crcFor = append(f.crcFor, crcFor)
@@ -103,7 +103,17 @@ func (f *frameSpec) prefill(buf []byte, patIdx int) {
copy(buf[minFrame:], f.refs[patIdx]) copy(buf[minFrame:], f.refs[patIdx])
} }
func putHeader(buf []byte, patIdx int, stream uint16, seq uint64, payLen int, crc uint32) { // Looked up rather than read from the frame: a checksum travelling beside the
// bytes it covers is only ever compared against itself.
func (f *frameSpec) expectedCRC(patIdx, payLen int) (uint32, bool) {
if patIdx < 0 || patIdx >= len(f.crcFor) {
return 0, false
}
crc, ok := f.crcFor[patIdx][payLen]
return crc, ok
}
func putHeader(buf []byte, patIdx int, stream uint16, seq uint64, payLen int) {
h := buf[ethHdrLen:] h := buf[ethHdrLen:]
binary.BigEndian.PutUint32(h[0:4], hdrMagic) binary.BigEndian.PutUint32(h[0:4], hdrMagic)
h[4] = 1 h[4] = 1
@@ -112,7 +122,6 @@ func putHeader(buf []byte, patIdx int, stream uint16, seq uint64, payLen int, cr
binary.BigEndian.PutUint64(h[8:16], seq) binary.BigEndian.PutUint64(h[8:16], seq)
binary.BigEndian.PutUint16(h[16:18], uint16(payLen)) binary.BigEndian.PutUint16(h[16:18], uint16(payLen))
binary.BigEndian.PutUint16(h[18:20], 0) binary.BigEndian.PutUint16(h[18:20], 0)
binary.BigEndian.PutUint32(h[20:24], crc)
} }
type parsed struct { type parsed struct {
@@ -120,7 +129,6 @@ type parsed struct {
stream uint16 stream uint16
seq uint64 seq uint64
payLen int payLen int
crc uint32
} }
func parseHeader(buf []byte) (parsed, bool) { func parseHeader(buf []byte) (parsed, bool) {
@@ -136,7 +144,6 @@ func parseHeader(buf []byte) (parsed, bool) {
p.stream = binary.BigEndian.Uint16(h[6:8]) p.stream = binary.BigEndian.Uint16(h[6:8])
p.seq = binary.BigEndian.Uint64(h[8:16]) p.seq = binary.BigEndian.Uint64(h[8:16])
p.payLen = int(binary.BigEndian.Uint16(h[16:18])) p.payLen = int(binary.BigEndian.Uint16(h[16:18]))
p.crc = binary.BigEndian.Uint32(h[20:24])
if minFrame+p.payLen > len(buf) { if minFrame+p.payLen > len(buf) {
return p, false return p, false
} }
+9 -12
View File
@@ -197,7 +197,7 @@ type sample struct {
lost, late uint64 lost, late uint64
crcErr, badMagic uint64 crcErr, badMagic uint64
badLen uint64 badLen uint64
txErrs, txShort uint64 txErrs uint64
rxErrs uint64 rxErrs uint64
} }
@@ -246,7 +246,6 @@ func (d *direction) snapshot() sample {
s.txFrames += t.frames.Load() s.txFrames += t.frames.Load()
s.txBytes += t.bytes.Load() s.txBytes += t.bytes.Load()
s.txErrs += t.errs.Load() s.txErrs += t.errs.Load()
s.txShort += t.short.Load()
} }
for _, r := range d.rxStats { for _, r := range d.rxStats {
s.rxFrames += r.frames.Load() s.rxFrames += r.frames.Load()
@@ -331,16 +330,14 @@ func errsBetween(b, n counterSet) errs {
// checksum, a header that is not ours, and a length that cannot be. // 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) + corrupt: (n.s.crcErr - b.s.crcErr) + (n.s.badMagic - b.s.badMagic) +
(n.s.badLen - b.s.badLen), (n.s.badLen - b.s.badLen),
// A frame the stack refused and a frame the driver dropped are the same // What the hardware reported. Nothing the host declined to send is here,
// failure seen from either side of the ring, and never the same frame // so this one going red means the cable.
// twice: a send that fails never reaches the driver to be dropped. link: (n.nic - b.nic) + (n.s.rxErrs - b.s.rxErrs),
link: (n.nic - b.nic) + (n.s.txErrs - b.s.txErrs) + (n.s.rxErrs - b.s.rxErrs), // Ours rather than the cable's. A late frame is unreachable while each
// Ours rather than the cable's: frames the kernel threw away because we // stream has a flow rule to its own queue, which is exactly why it is
// did not drain the ring fast enough, and frames so far out of order // worth counting.
// that our own bookkeeping had already written them off. The second is internal: (n.drops - b.drops) + (n.s.late - b.s.late) +
// unreachable while each stream has a flow rule to its own queue, which (n.s.txErrs - b.s.txErrs),
// is exactly why it is worth counting: it fires if that stops holding.
internal: (n.drops - b.drops) + (n.s.late - b.s.late),
} }
} }
+1 -2
View File
@@ -156,8 +156,7 @@ func (p *probeSender) run(done *atomic.Bool, startTx <-chan struct{}) {
} }
} }
putHeader(buf, probePattern, probeStream, seq, probeSize-minFrame, putHeader(buf, probePattern, probeStream, seq, probeSize-minFrame)
p.spec.crcFor[probePattern][probeSize])
err := unix.Send(p.fd, buf, 0) err := unix.Send(p.fd, buf, 0)
// The sequence advances even when a probe fails, so a stale receive half // The sequence advances even when a probe fails, so a stale receive half
// can never be paired with a later probe that reused its number. // can never be paired with a later probe that reused its number.
+3 -2
View File
@@ -61,12 +61,13 @@ func (w *rxWorker) run(done *atomic.Bool) {
w.streams[p.stream].observe(p.seq) w.streams[p.stream].observe(p.seq)
} }
if p.payLen > w.spec.maxPay { want, ok := w.spec.expectedCRC(p.patIdx, p.payLen)
if !ok {
w.stats.badLen.Add(1) w.stats.badLen.Add(1)
continue continue
} }
pay := buf[minFrame : minFrame+p.payLen] pay := buf[minFrame : minFrame+p.payLen]
if crc32.Checksum(pay, crcTable) != p.crc { if crc32.Checksum(pay, crcTable) != want {
w.stats.crcErr.Add(1) w.stats.crcErr.Add(1)
} }
} }
+5 -10
View File
@@ -10,8 +10,7 @@ type txStats struct {
frames atomic.Uint64 frames atomic.Uint64
bytes atomic.Uint64 bytes atomic.Uint64
errs atomic.Uint64 errs atomic.Uint64
short atomic.Uint64 _ [40]byte
_ [32]byte
} }
type txWorker struct { type txWorker struct {
@@ -48,8 +47,7 @@ func (w *txWorker) run(done *atomic.Bool) {
si = 0 si = 0
} }
sizes[i] = size sizes[i] = size
putHeader(bufs[i], pats[i], w.stream, seq+uint64(i), size-minFrame, putHeader(bufs[i], pats[i], w.stream, seq+uint64(i), size-minFrame)
w.spec.crcFor[pats[i]][size])
iovs[i].Len = uint64(size) iovs[i].Len = uint64(size)
} }
@@ -63,13 +61,10 @@ func (w *txWorker) run(done *atomic.Bool) {
w.stats.bytes.Add(b) w.stats.bytes.Add(b)
seq += uint64(n) seq += uint64(n)
} }
if n < w.batch { // Taking fewer of the vector than offered is the ring's room, not a frame
switch { // lost: the rest go on the next pass.
case n < 0 && err != unix.EINTR && err != unix.EAGAIN && err != unix.ENOBUFS: if n < 0 && err != unix.EINTR && err != unix.EAGAIN && err != unix.ENOBUFS {
w.stats.errs.Add(1) w.stats.errs.Add(1)
default:
w.stats.short.Add(1)
}
} }
} }
} }