package main

import (
	"hash/crc32"
	"sync"
	"sync/atomic"

	"golang.org/x/sys/unix"
)

type rxStats struct {
	frames   atomic.Uint64
	bytes    atomic.Uint64
	badMagic atomic.Uint64
	badHdr   atomic.Uint64
	badLen   atomic.Uint64
	crcErr   atomic.Uint64
	rxErrs   atomic.Uint64

	// Frames counted into the interval their mac receive stamp falls in, rather
	// than the interval a worker got round to draining them in.
	newest  atomic.Int64
	buckets [rateBuckets]rxBucket
}

// One sample interval of arrivals, keyed by the mac's clock, with enough of them
// kept that a bucket is read long before its slot comes round again.
const (
	rateBucketNs   = int64(sampleInterval)
	rateBuckets    = 64
	rateBucketSecs = float64(rateBucketNs) / 1e9
)

type rxBucket struct {
	epoch  atomic.Int64
	frames atomic.Uint64
	bytes  atomic.Uint64
}

// Only the owning worker writes its own buckets, so a slot coming round again is
// simply zeroed before it is claimed for the new epoch.
func (s *rxStats) observe(stamp int64, n uint64) {
	e := stamp / rateBucketNs
	b := &s.buckets[e&(rateBuckets-1)]
	if b.epoch.Load() != e {
		b.frames.Store(0)
		b.bytes.Store(0)
		b.epoch.Store(e)
	}
	b.frames.Add(1)
	b.bytes.Add(n)
	if e > s.newest.Load() {
		s.newest.Store(e)
	}
}

// What this worker counted into one epoch, or nothing if that epoch has already
// fallen out of the ring.
func (s *rxStats) bucket(e int64) (frames, bytes uint64) {
	b := &s.buckets[e&(rateBuckets-1)]
	if b.epoch.Load() != e {
		return 0, 0
	}
	return b.frames.Load(), b.bytes.Load()
}

type rxWorker struct {
	fd     int
	batch  int
	stream uint16
	spec   *frameSpec
	stats  *rxStats
	loss   *lossWindow
	ready  *sync.WaitGroup
}

func (w *rxWorker) run(done *atomic.Bool) {
	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, oob := newRxMmsghdrs(bufs)

	w.ready.Done()

	for !done.Load() {
		// The kernel overwrites each Controllen with what it wrote, so they are
		// reset before every call.
		for i := range hdrs {
			hdrs[i].hdr.Controllen = cmsgLen
		}
		n, err := recvmmsg(w.fd, hdrs, unix.MSG_WAITFORONE)
		if n <= 0 {
			if err != nil && err != unix.EAGAIN && err != unix.EINTR {
				w.stats.rxErrs.Add(1)
			}
			continue
		}
		for i := 0; i < n; i++ {
			buf := bufs[i][:int(hdrs[i].len)]
			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)
			w.stats.bytes.Add(uint64(len(buf)))
			if ts, ok := hwTimestamp(oob[i][:hdrs[i].hdr.Controllen]); ok {
				w.stats.observe(ts, uint64(len(buf)))
			}

			// 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
			}

			want, ok := w.spec.expectedCRC(p.patIdx, p.payLen)
			if !ok {
				w.stats.badLen.Add(1)
				continue
			}
			pay := buf[minFrame : minFrame+p.payLen]
			if crc32.Checksum(pay, crcTable) != want {
				w.stats.crcErr.Add(1)
			}
		}
	}
}