cabletest/tx.go

package main

import (
	"sync/atomic"

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

type txStats struct {
	errs atomic.Uint64

	// The exclusive upper bound on any sequence number this stream can have put
	// on the wire. Read by the receiving half of the same stream, which is the
	// only thing that can tell a gap it must write off from a sequence number
	// that was never sent at all.
	sent atomic.Uint64
}

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)

	<-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
			}
			putHeader(bufs[i], pats[i], w.stream, seq+uint64(i), size-minFrame)
			iovs[i].Len = uint64(size)
		}

		// Published before the send rather than after it, so it is never behind a
		// frame already in flight. Overshooting a partial send is harmless: it is
		// a bound, and one the next pass raises again.
		w.stats.sent.Store(seq + uint64(w.batch))

		n, err := sendmmsg(w.fd, hdrs)
		if n > 0 {
			seq += uint64(n)
		}
		// Taking fewer of the vector than offered is the ring's room, not a frame
		// lost: the rest go on the next pass.
		if n < 0 && err != unix.EINTR && err != unix.EAGAIN && err != unix.ENOBUFS {
			w.stats.errs.Add(1)
		}
	}
}