package main

import (
	"sync/atomic"

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

type txStats struct {
	frames atomic.Uint64
	bytes  atomic.Uint64
	errs   atomic.Uint64
	short  atomic.Uint64
	_      [32]byte
}

type txWorker struct {
	fd      int
	stream  uint16
	spec    *frameSpec
	batch   int
	stats   *txStats
	startTx <-chan struct{}
}

func (w *txWorker) run(done *atomic.Bool) {
	// The batch is not a multiple of the pattern count, so the pairing of pattern
	// to size rotates every pass rather than settling into a fixed one.
	bufs := make([][]byte, w.batch)
	pats := make([]int, w.batch)
	for i := range bufs {
		bufs[i] = make([]byte, w.spec.maxSize)
		pats[i] = i % len(patterns)
		w.spec.prefill(bufs[i], pats[i])
	}
	hdrs, iovs := newMmsghdrs(bufs)
	sizes := make([]int, w.batch)

	<-w.startTx

	var seq uint64
	si := 0
	for !done.Load() {
		for i := 0; i < w.batch; i++ {
			size := w.spec.sizes[si]
			si++
			if si == len(w.spec.sizes) {
				si = 0
			}
			sizes[i] = size
			putHeader(bufs[i], pats[i], w.stream, seq+uint64(i), size-minFrame)
			iovs[i].Len = uint64(size)
		}

		n, err := sendmmsg(w.fd, hdrs)
		if n > 0 {
			var b uint64
			for i := 0; i < n; i++ {
				b += uint64(sizes[i])
			}
			w.stats.frames.Add(uint64(n))
			w.stats.bytes.Add(b)
			seq += uint64(n)
		}
		if n < w.batch {
			switch {
			case n < 0 && err != unix.EINTR && err != unix.EAGAIN && err != unix.ENOBUFS:
				w.stats.errs.Add(1)
			default:
				w.stats.short.Add(1)
			}
		}
	}
}