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) { bufs := make([][]byte, w.batch) for i := range bufs { bufs[i] = make([]byte, w.spec.maxSize) w.spec.prefill(bufs[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], w.spec.patIdx, w.stream, seq+uint64(i), size-minFrame, w.spec.crcFor[size]) 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) } } } }