103 lines
1.9 KiB
Go
103 lines
1.9 KiB
Go
|
package main
|
||
|
|
|
||
|
|
import (
|
||
|
|
"fmt"
|
||
|
|
"hash/crc32"
|
||
|
|
"sync"
|
||
|
|
"sync/atomic"
|
||
|
|
|
||
|
|
"golang.org/x/sys/unix"
|
||
|
|
)
|
||
|
|
|
||
|
|
type rxStats struct {
|
||
|
|
frames atomic.Uint64
|
||
|
|
bytes atomic.Uint64
|
||
|
|
badMagic atomic.Uint64
|
||
|
|
badLen atomic.Uint64
|
||
|
|
crcErr atomic.Uint64
|
||
|
|
_ [24]byte
|
||
|
|
}
|
||
|
|
|
||
|
|
type streamState struct {
|
||
|
|
maxSeq atomic.Uint64
|
||
|
|
count atomic.Uint64
|
||
|
|
_ [48]byte
|
||
|
|
}
|
||
|
|
|
||
|
|
type rxWorker struct {
|
||
|
|
fd int
|
||
|
|
batch int
|
||
|
|
cpu int
|
||
|
|
spec *frameSpec
|
||
|
|
stats *rxStats
|
||
|
|
streams []streamState
|
||
|
|
reports chan string
|
||
|
|
ready *sync.WaitGroup
|
||
|
|
}
|
||
|
|
|
||
|
|
func (w *rxWorker) run(done *atomic.Bool) {
|
||
|
|
pinTo(w.cpu)
|
||
|
|
|
||
|
|
bufs := make([][]byte, w.batch)
|
||
|
|
for i := range bufs {
|
||
|
|
bufs[i] = make([]byte, maxFrame)
|
||
|
|
for j := 0; j < maxFrame; j += 4096 {
|
||
|
|
bufs[i][j] = 0
|
||
|
|
}
|
||
|
|
}
|
||
|
|
hdrs, _ := newMmsghdrs(bufs)
|
||
|
|
|
||
|
|
w.ready.Done()
|
||
|
|
|
||
|
|
for !done.Load() {
|
||
|
|
n, err := recvmmsg(w.fd, hdrs, unix.MSG_WAITFORONE)
|
||
|
|
if n <= 0 {
|
||
|
|
if err != nil && err != unix.EAGAIN && err != unix.EINTR {
|
||
|
|
w.report(fmt.Sprintf("recvmmsg: %v", err))
|
||
|
|
}
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
for i := 0; i < n; i++ {
|
||
|
|
buf := bufs[i][:int(hdrs[i].len)]
|
||
|
|
p, ok := parseHeader(buf)
|
||
|
|
if !ok {
|
||
|
|
w.stats.badMagic.Add(1)
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
w.stats.frames.Add(1)
|
||
|
|
w.stats.bytes.Add(uint64(len(buf)))
|
||
|
|
|
||
|
|
if int(p.stream) < len(w.streams) {
|
||
|
|
st := &w.streams[p.stream]
|
||
|
|
st.count.Add(1)
|
||
|
|
for {
|
||
|
|
old := st.maxSeq.Load()
|
||
|
|
if p.seq <= old || st.maxSeq.CompareAndSwap(old, p.seq) {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
if p.payLen > len(w.spec.ref) {
|
||
|
|
w.stats.badLen.Add(1)
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
pay := buf[minFrame : minFrame+p.payLen]
|
||
|
|
if crc32.Checksum(pay, crcTable) == p.crc {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
w.stats.crcErr.Add(1)
|
||
|
|
off, bits := firstDiff(pay, w.spec.ref[:p.payLen])
|
||
|
|
w.report(fmt.Sprintf("payload mismatch stream=%d seq=%d len=%d first-diff-offset=%d bits=%d",
|
||
|
|
p.stream, p.seq, p.payLen, off, bits))
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (w *rxWorker) report(msg string) {
|
||
|
|
select {
|
||
|
|
case w.reports <- msg:
|
||
|
|
default:
|
||
|
|
}
|
||
|
|
}
|