diff --git a/main.go b/main.go index a0e0daf..37eeb3f 100644 --- a/main.go +++ b/main.go @@ -464,12 +464,13 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, rxRea } for i, fd := range d.rxFDs { w := &rxWorker{ - fd: fd, - batch: batchSize, - spec: d.specs[i], - stats: d.rxStats[i], - streams: d.streams, - ready: rxReady, + fd: fd, + batch: batchSize, + stream: uint16(i), + spec: d.specs[i], + stats: d.rxStats[i], + loss: &d.streams[i], + ready: rxReady, } wg.Add(1) go func() { diff --git a/rx.go b/rx.go index 902d876..9f9e6c2 100644 --- a/rx.go +++ b/rx.go @@ -65,12 +65,13 @@ func (s *rxStats) bucket(e int64) (frames, bytes uint64) { } type rxWorker struct { - fd int - batch int - spec *frameSpec - stats *rxStats - streams []lossWindow - ready *sync.WaitGroup + fd int + batch int + stream uint16 + spec *frameSpec + stats *rxStats + loss *lossWindow + ready *sync.WaitGroup } func (w *rxWorker) run(done *atomic.Bool) { @@ -115,11 +116,10 @@ func (w *rxWorker) run(done *atomic.Bool) { w.stats.observe(ts, uint64(len(buf))) } - // 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) { + // The ethertype this socket is bound to already says which stream the + // frame belongs to, so a header naming another one is damaged, as is a + // sequence number the sender never reached. + if p.stream != w.stream || !w.loss.observe(p.seq) { w.stats.badHdr.Add(1) continue }