From befd732406cc0e8033d9210664695f2c8e14a3e3 Mon Sep 17 00:00:00 2001 From: flamingcow Date: Sat, 25 Jul 2026 17:58:54 -0700 Subject: [PATCH] Add full-duplex raw-Ethernet 10G cable stress tester --- counters.go | 71 ++++++++ frame.go | 175 ++++++++++++++++++ go.mod | 2 + go.sum | 2 + main.go | 506 +++++++++++++++++++++++++++++++++++++++++++++++++++- rx.go | 106 +++++++++++ sock.go | 137 ++++++++++++++ tx.go | 73 ++++++++ 8 files changed, 1070 insertions(+), 2 deletions(-) create mode 100644 counters.go create mode 100644 frame.go create mode 100644 go.sum create mode 100644 rx.go create mode 100644 sock.go create mode 100644 tx.go diff --git a/counters.go b/counters.go new file mode 100644 index 0000000..d287996 --- /dev/null +++ b/counters.go @@ -0,0 +1,71 @@ +package main + +import ( + "fmt" + "os" + "strconv" + "strings" +) + +var nicFields = []string{ + "rx_packets", "tx_packets", + "rx_errors", "tx_errors", + "rx_dropped", "tx_dropped", + "rx_crc_errors", "rx_missed_errors", + "rx_length_errors", "rx_over_errors", + "rx_frame_errors", "rx_fifo_errors", + "collisions", +} + +type nicCounters struct { + stats map[string]uint64 + carrierChanges uint64 + carrierUp uint64 + carrierDown uint64 +} + +func readUint(path string) (uint64, bool) { + b, err := os.ReadFile(path) + if err != nil { + return 0, false + } + v, err := strconv.ParseUint(strings.TrimSpace(string(b)), 10, 64) + if err != nil { + return 0, false + } + return v, true +} + +func readNIC(ifname string) nicCounters { + c := nicCounters{stats: make(map[string]uint64, len(nicFields))} + base := "/sys/class/net/" + ifname + for _, f := range nicFields { + if v, ok := readUint(base + "/statistics/" + f); ok { + c.stats[f] = v + } + } + c.carrierChanges, _ = readUint(base + "/carrier_changes") + c.carrierUp, _ = readUint(base + "/carrier_up_count") + c.carrierDown, _ = readUint(base + "/carrier_down_count") + return c +} + +func (c nicCounters) diff(prev nicCounters) string { + var parts []string + for _, f := range nicFields { + now, ok := c.stats[f] + if !ok { + continue + } + if d := now - prev.stats[f]; d != 0 && !strings.HasSuffix(f, "_packets") { + parts = append(parts, fmt.Sprintf("%s=%d", f, d)) + } + } + if d := c.carrierChanges - prev.carrierChanges; d != 0 { + parts = append(parts, fmt.Sprintf("carrier_changes=%d", d)) + } + if d := c.carrierDown - prev.carrierDown; d != 0 { + parts = append(parts, fmt.Sprintf("carrier_down=%d", d)) + } + return strings.Join(parts, " ") +} diff --git a/frame.go b/frame.go new file mode 100644 index 0000000..3090c8f --- /dev/null +++ b/frame.go @@ -0,0 +1,175 @@ +package main + +import ( + "encoding/binary" + "fmt" + "hash/crc32" + "math/rand/v2" +) + +const ( + etherType = 0x88b5 + ethHdrLen = 14 + hdrLen = 24 + hdrMagic = 0x43424c54 + minFrame = ethHdrLen + hdrLen + maxFrame = 9216 +) + +var crcTable = crc32.MakeTable(crc32.Castagnoli) + +type pattern struct { + name string + fill func([]byte) +} + +var patterns = []pattern{ + {"zeros", func(b []byte) { clear(b) }}, + {"ones", func(b []byte) { + for i := range b { + b[i] = 0xff + } + }}, + {"alt", func(b []byte) { + for i := range b { + if i%2 == 0 { + b[i] = 0x55 + } else { + b[i] = 0xaa + } + } + }}, + {"incr", func(b []byte) { + for i := range b { + b[i] = byte(i) + } + }}, + {"prbs", func(b []byte) { + s := uint64(0x0123456789abcdef) + for i := range b { + s ^= s << 13 + s ^= s >> 7 + s ^= s << 17 + b[i] = byte(s) + } + }}, + {"random", func(b []byte) { + r := rand.New(rand.NewPCG(0xc0ffee, 0x5eed)) + for i := range b { + b[i] = byte(r.Uint32()) + } + }}, +} + +func patternIndex(name string) (int, error) { + for i, p := range patterns { + if p.name == name { + return i, nil + } + } + names := make([]string, len(patterns)) + for i, p := range patterns { + names[i] = p.name + } + return 0, fmt.Errorf("unknown pattern %q (have %v)", name, names) +} + +type frameSpec struct { + patIdx int + ref []byte + crcFor map[int]uint32 + dstMAC [6]byte + srcMAC [6]byte + sizes []int + maxSize int +} + +func newFrameSpec(patIdx int, dst, src [6]byte, sizes []int) *frameSpec { + maxSize := 0 + for _, s := range sizes { + if s > maxSize { + maxSize = s + } + } + ref := make([]byte, maxSize-minFrame) + patterns[patIdx].fill(ref) + crcFor := make(map[int]uint32, len(sizes)) + for _, s := range sizes { + crcFor[s] = crc32.Checksum(ref[:s-minFrame], crcTable) + } + return &frameSpec{ + patIdx: patIdx, + ref: ref, + crcFor: crcFor, + dstMAC: dst, + srcMAC: src, + sizes: sizes, + maxSize: maxSize, + } +} + +func (f *frameSpec) prefill(buf []byte) { + copy(buf[0:6], f.dstMAC[:]) + copy(buf[6:12], f.srcMAC[:]) + binary.BigEndian.PutUint16(buf[12:14], etherType) + copy(buf[minFrame:], f.ref) +} + +func putHeader(buf []byte, patIdx int, stream uint16, seq uint64, payLen int, crc uint32) { + h := buf[ethHdrLen:] + binary.BigEndian.PutUint32(h[0:4], hdrMagic) + h[4] = 1 + h[5] = byte(patIdx) + binary.BigEndian.PutUint16(h[6:8], stream) + binary.BigEndian.PutUint64(h[8:16], seq) + binary.BigEndian.PutUint16(h[16:18], uint16(payLen)) + binary.BigEndian.PutUint16(h[18:20], 0) + binary.BigEndian.PutUint32(h[20:24], crc) +} + +type parsed struct { + patIdx int + stream uint16 + seq uint64 + payLen int + crc uint32 +} + +func parseHeader(buf []byte) (parsed, bool) { + var p parsed + if len(buf) < minFrame { + return p, false + } + h := buf[ethHdrLen:] + if binary.BigEndian.Uint32(h[0:4]) != hdrMagic { + return p, false + } + p.patIdx = int(h[5]) + p.stream = binary.BigEndian.Uint16(h[6:8]) + p.seq = binary.BigEndian.Uint64(h[8:16]) + p.payLen = int(binary.BigEndian.Uint16(h[16:18])) + p.crc = binary.BigEndian.Uint32(h[20:24]) + if minFrame+p.payLen > len(buf) { + return p, false + } + return p, true +} + +func firstDiff(got, want []byte) (int, int) { + n := len(got) + if len(want) < n { + n = len(want) + } + for i := 0; i < n; i++ { + if got[i] != want[i] { + bits := 0 + x := got[i] ^ want[i] + for x != 0 { + bits += int(x & 1) + x >>= 1 + } + return i, bits + } + } + return -1, 0 +} diff --git a/go.mod b/go.mod index 5b9a84d..de6d7c1 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,5 @@ module g.fc.run/theater/cabletest go 1.26.5 + +require golang.org/x/sys v0.47.0 // indirect diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..37ee2d4 --- /dev/null +++ b/go.sum @@ -0,0 +1,2 @@ +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= diff --git a/main.go b/main.go index f7b60bd..150d259 100644 --- a/main.go +++ b/main.go @@ -1,7 +1,509 @@ package main -import "fmt" +import ( + "flag" + "fmt" + "net" + "os" + "os/signal" + "strconv" + "strings" + "sync" + "sync/atomic" + "syscall" + "time" + + "golang.org/x/sys/unix" +) + +const ( + fanoutHash = 0 + fanoutLB = 1 + fanoutCPU = 2 + fanoutRollover = 3 +) + +const wireOverhead = 24 + +type endpoint struct { + name string + idx int + mac [6]byte + mtu int +} + +type direction struct { + label string + tx endpoint + rx endpoint + spec *frameSpec + txStats []*txStats + rxStats []*rxStats + streams []streamState + txFDs []int + rxFDs []int + reports chan string + + prev sample + drops uint64 + nicTX nicCounters + nicRX nicCounters +} + +type sample struct { + txFrames, txBytes uint64 + rxFrames, rxBytes uint64 + expected, count uint64 + crcErr, badMagic uint64 + badLen uint64 + txErrs, txShort uint64 +} + +func lookupEndpoint(name string) (endpoint, error) { + ifi, err := net.InterfaceByName(name) + if err != nil { + return endpoint{}, err + } + if len(ifi.HardwareAddr) != 6 { + return endpoint{}, fmt.Errorf("%s: expected 6-byte MAC, got %q", name, ifi.HardwareAddr) + } + var mac [6]byte + copy(mac[:], ifi.HardwareAddr) + return endpoint{name: name, idx: ifi.Index, mac: mac, mtu: ifi.MTU}, nil +} + +func parseSizes(s string) ([]int, error) { + var out []int + for _, f := range strings.Split(s, ",") { + f = strings.TrimSpace(f) + if f == "" { + continue + } + v, err := strconv.Atoi(f) + if err != nil { + return nil, fmt.Errorf("bad size %q: %w", f, err) + } + if v < minFrame { + return nil, fmt.Errorf("size %d below minimum %d", v, minFrame) + } + out = append(out, v) + } + if len(out) == 0 { + return nil, fmt.Errorf("no sizes given") + } + return out, nil +} + +func parseCPUs(s string) ([]int, error) { + if strings.TrimSpace(s) == "" { + return nil, nil + } + var out []int + for _, f := range strings.Split(s, ",") { + v, err := strconv.Atoi(strings.TrimSpace(f)) + if err != nil { + return nil, fmt.Errorf("bad cpu %q: %w", f, err) + } + out = append(out, v) + } + return out, nil +} + +func cpuAt(cpus []int, i int) int { + if len(cpus) == 0 { + return -1 + } + return cpus[i%len(cpus)] +} + +func fanoutMode(name string, nsock int) (int, error) { + if nsock < 2 { + return -1, nil + } + switch name { + case "none": + return -1, nil + case "hash": + return fanoutHash, nil + case "lb": + return fanoutLB, nil + case "cpu": + return fanoutCPU, nil + case "rollover": + return fanoutRollover, nil + } + return 0, fmt.Errorf("unknown fanout mode %q", name) +} + +func (d *direction) snapshot() sample { + var s sample + for _, t := range d.txStats { + s.txFrames += t.frames.Load() + s.txBytes += t.bytes.Load() + s.txErrs += t.errs.Load() + s.txShort += t.short.Load() + } + for _, r := range d.rxStats { + s.rxFrames += r.frames.Load() + s.rxBytes += r.bytes.Load() + s.crcErr += r.crcErr.Load() + s.badMagic += r.badMagic.Load() + s.badLen += r.badLen.Load() + } + for i := range d.streams { + c := d.streams[i].count.Load() + if c == 0 { + continue + } + s.count += c + s.expected += d.streams[i].maxSeq.Load() + 1 + } + return s +} + +func (d *direction) sampleDrops() { + for _, fd := range d.rxFDs { + d.drops += packetDrops(fd) + } +} + +func gbps(bytes, frames uint64, secs float64) float64 { + return float64((bytes+frames*wireOverhead)*8) / secs / 1e9 +} + +func (d *direction) reportInterval(secs float64) string { + now := d.snapshot() + p := d.prev + d.prev = now + + txF := now.txFrames - p.txFrames + txB := now.txBytes - p.txBytes + rxF := now.rxFrames - p.rxFrames + rxB := now.rxBytes - p.rxBytes + + lost := int64(now.expected) - int64(now.count) + + before := d.drops + d.sampleDrops() + drops := d.drops - before + + return fmt.Sprintf("%s tx %8.0f pps %6.2f Gb/s | rx %8.0f pps %6.2f Gb/s | lost(cum) %6d crc %d badmagic %d kdrop %d txshort %d txerr %d", + d.label, + float64(txF)/secs, gbps(txB, txF, secs), + float64(rxF)/secs, gbps(rxB, rxF, secs), + lost, + now.crcErr-p.crcErr, + now.badMagic-p.badMagic, + drops, + now.txShort-p.txShort, + now.txErrs-p.txErrs, + ) +} + +func (d *direction) reportNIC() string { + tx := readNIC(d.tx.name) + rx := readNIC(d.rx.name) + var parts []string + if s := tx.diff(d.nicTX); s != "" { + parts = append(parts, " "+d.tx.name+"(tx): "+s) + } + if s := rx.diff(d.nicRX); s != "" { + parts = append(parts, " "+d.rx.name+"(rx): "+s) + } + d.nicTX = tx + d.nicRX = rx + return strings.Join(parts, "\n") +} + +func buildDirection(label string, tx, rx endpoint, patIdx int, sizes []int, cfg config) (*direction, error) { + d := &direction{ + label: label, + tx: tx, + rx: rx, + spec: newFrameSpec(patIdx, rx.mac, tx.mac, sizes), + streams: make([]streamState, cfg.txWorkers), + reports: make(chan string, 64), + } + d.nicTX = readNIC(tx.name) + d.nicRX = readNIC(rx.name) + + fm, err := fanoutMode(cfg.fanout, cfg.rxWorkers) + if err != nil { + return nil, err + } + + for i := 0; i < cfg.txWorkers; i++ { + fd, err := openTxSocket(tx.idx, cfg.sndbuf) + if err != nil { + return nil, fmt.Errorf("%s tx socket: %w", label, err) + } + d.txFDs = append(d.txFDs, fd) + d.txStats = append(d.txStats, &txStats{}) + } + for i := 0; i < cfg.rxWorkers; i++ { + fd, err := openRxSocket(rx.idx, cfg.rcvbuf, cfg.fanoutID, fm) + if err != nil { + return nil, fmt.Errorf("%s rx socket: %w", label, err) + } + d.rxFDs = append(d.rxFDs, fd) + d.rxStats = append(d.rxStats, &rxStats{}) + } + return d, nil +} + +func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, cfg config, rxReady *sync.WaitGroup, startTx <-chan struct{}) { + for i, fd := range d.txFDs { + w := &txWorker{ + fd: fd, + stream: uint16(i), + spec: d.spec, + batch: cfg.batch, + cpu: cpuAt(cfg.txCPUs, i), + stats: d.txStats[i], + startTx: startTx, + } + wg.Add(1) + go func() { + defer wg.Done() + w.run(doneTx) + }() + } + for i, fd := range d.rxFDs { + w := &rxWorker{ + fd: fd, + batch: cfg.batch, + cpu: cpuAt(cfg.rxCPUs, i), + verify: cfg.verify, + spec: d.spec, + stats: d.rxStats[i], + streams: d.streams, + reports: d.reports, + ready: rxReady, + } + wg.Add(1) + go func() { + defer wg.Done() + w.run(doneRx) + }() + } +} + +func (d *direction) close() { + for _, fd := range d.txFDs { + unix.Close(fd) + } + for _, fd := range d.rxFDs { + unix.Close(fd) + } +} + +type config struct { + txWorkers int + rxWorkers int + batch int + verify bool + sndbuf int + rcvbuf int + fanout string + fanoutID int + txCPUs []int + rxCPUs []int +} func main() { - fmt.Println("Hello, world!") + var ( + aName = flag.String("a", "", "first interface") + bName = flag.String("b", "", "second interface") + duration = flag.Duration("duration", 0, "run time, 0 for until interrupted") + sizesArg = flag.String("sizes", "64,128,256,512,1024,1280,1514", "frame sizes in bytes, excluding FCS, cycled per packet") + patArg = flag.String("pattern", "prbs", "payload pattern") + txN = flag.Int("tx", 4, "tx workers per direction") + rxN = flag.Int("rx", 4, "rx workers per direction") + batch = flag.Int("batch", 64, "frames per sendmmsg/recvmmsg call") + verify = flag.Bool("verify", true, "verify payload CRC32C on receive") + interval = flag.Duration("interval", time.Second, "report interval") + drain = flag.Duration("drain", 500*time.Millisecond, "keep receiving this long after tx stops, so in-flight frames are not counted as lost") + fanout = flag.String("fanout", "lb", "rx fanout mode: none, hash, lb, cpu, rollover") + duplex = flag.Bool("duplex", true, "run both directions simultaneously") + sndbuf = flag.Int("sndbuf", 8<<20, "SO_SNDBUFFORCE per tx socket") + rcvbuf = flag.Int("rcvbuf", 64<<20, "SO_RCVBUFFORCE per rx socket") + txCPUs = flag.String("txcpus", "", "comma-separated CPUs to pin tx workers to") + rxCPUs = flag.String("rxcpus", "", "comma-separated CPUs to pin rx workers to") + ) + flag.Parse() + + if err := run(*aName, *bName, *sizesArg, *patArg, *fanout, *txCPUs, *rxCPUs, + *duration, *interval, *drain, *txN, *rxN, *batch, *sndbuf, *rcvbuf, *verify, *duplex); err != nil { + fmt.Fprintln(os.Stderr, "error:", err) + os.Exit(1) + } +} + +func run(aName, bName, sizesArg, patArg, fanout, txCPUsArg, rxCPUsArg string, + duration, interval, drain time.Duration, txN, rxN, batch, sndbuf, rcvbuf int, + verify, duplex bool) error { + + if aName == "" || bName == "" { + return fmt.Errorf("both -a and -b are required") + } + sizes, err := parseSizes(sizesArg) + if err != nil { + return err + } + patIdx, err := patternIndex(patArg) + if err != nil { + return err + } + txCPUs, err := parseCPUs(txCPUsArg) + if err != nil { + return err + } + rxCPUs, err := parseCPUs(rxCPUsArg) + if err != nil { + return err + } + a, err := lookupEndpoint(aName) + if err != nil { + return err + } + b, err := lookupEndpoint(bName) + if err != nil { + return err + } + for _, e := range []endpoint{a, b} { + for _, s := range sizes { + if s > e.mtu+ethHdrLen { + return fmt.Errorf("size %d exceeds %s MTU %d (max frame %d)", s, e.name, e.mtu, e.mtu+ethHdrLen) + } + } + } + + cfg := config{ + txWorkers: txN, + rxWorkers: rxN, + batch: batch, + verify: verify, + sndbuf: sndbuf, + rcvbuf: rcvbuf, + fanout: fanout, + txCPUs: txCPUs, + rxCPUs: rxCPUs, + } + + var dirs []*direction + cfgA := cfg + cfgA.fanoutID = 0x4341 + d0, err := buildDirection(a.name+"->"+b.name, a, b, patIdx, sizes, cfgA) + if err != nil { + return err + } + dirs = append(dirs, d0) + if duplex { + cfgB := cfg + cfgB.fanoutID = 0x4342 + d1, err := buildDirection(b.name+"->"+a.name, b, a, patIdx, sizes, cfgB) + if err != nil { + return err + } + dirs = append(dirs, d1) + } + defer func() { + for _, d := range dirs { + d.close() + } + }() + + fmt.Printf("pattern=%s sizes=%v tx=%d rx=%d batch=%d verify=%v fanout=%s duplex=%v\n", + patterns[patIdx].name, sizes, txN, rxN, batch, verify, fanout, duplex) + for _, d := range dirs { + fmt.Printf(" %s %02x:%02x:%02x:%02x:%02x:%02x -> %02x:%02x:%02x:%02x:%02x:%02x ethertype 0x%04x\n", + d.label, + d.tx.mac[0], d.tx.mac[1], d.tx.mac[2], d.tx.mac[3], d.tx.mac[4], d.tx.mac[5], + d.rx.mac[0], d.rx.mac[1], d.rx.mac[2], d.rx.mac[3], d.rx.mac[4], d.rx.mac[5], + etherType) + } + fmt.Printf(" sndbuf %d rcvbuf %d (as granted by kernel)\n", + sockBufSize(dirs[0].txFDs[0], unix.SO_SNDBUF), + sockBufSize(dirs[0].rxFDs[0], unix.SO_RCVBUF)) + + var doneTx, doneRx atomic.Bool + var wg sync.WaitGroup + var rxReady sync.WaitGroup + startTx := make(chan struct{}) + for _, d := range dirs { + rxReady.Add(len(d.rxFDs)) + } + for _, d := range dirs { + d.start(&wg, &doneTx, &doneRx, cfg, &rxReady, startTx) + } + rxReady.Wait() + + sig := make(chan os.Signal, 1) + signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) + + start := time.Now() + close(startTx) + tick := time.NewTicker(interval) + defer tick.Stop() + + var deadline <-chan time.Time + if duration > 0 { + t := time.NewTimer(duration) + defer t.Stop() + deadline = t.C + } + + last := time.Now() +loop: + for { + select { + case <-sig: + break loop + case <-deadline: + break loop + case now := <-tick.C: + secs := now.Sub(last).Seconds() + last = now + for _, d := range dirs { + fmt.Println(d.reportInterval(secs)) + if s := d.reportNIC(); s != "" { + fmt.Println(s) + } + for { + select { + case msg := <-d.reports: + fmt.Println(" " + d.label + ": " + msg) + continue + default: + } + break + } + } + } + } + + doneTx.Store(true) + elapsed := time.Since(start).Seconds() + time.Sleep(drain) + doneRx.Store(true) + wg.Wait() + + fmt.Println("---") + for _, d := range dirs { + d.sampleDrops() + s := d.snapshot() + lost := int64(s.expected) - int64(s.count) + fmt.Printf("%s total tx %d frames %.2f GB | rx %d frames %.2f GB | lost %d (%.3g%%) crc %d badmagic %d badlen %d kdrop %d\n", + d.label, + s.txFrames, float64(s.txBytes)/1e9, + s.rxFrames, float64(s.rxBytes)/1e9, + lost, 100*float64(lost)/float64(max(s.expected, 1)), + s.crcErr, s.badMagic, s.badLen, d.drops) + fmt.Printf("%s avg tx %.2f Gb/s rx %.2f Gb/s over %.1fs\n", + d.label, + gbps(s.txBytes, s.txFrames, elapsed), + gbps(s.rxBytes, s.rxFrames, elapsed), + elapsed) + } + return nil } diff --git a/rx.go b/rx.go new file mode 100644 index 0000000..0cc505b --- /dev/null +++ b/rx.go @@ -0,0 +1,106 @@ +package main + +import ( + "fmt" + "hash/crc32" + "sync" + "sync/atomic" + + "golang.org/x/sys/unix" +) + +type rxStats struct { + frames atomic.Uint64 + bytes atomic.Uint64 + badMagic atomic.Uint64 + badLen atomic.Uint64 + crcErr atomic.Uint64 + _ [24]byte +} + +type streamState struct { + maxSeq atomic.Uint64 + count atomic.Uint64 + _ [48]byte +} + +type rxWorker struct { + fd int + batch int + cpu int + verify bool + spec *frameSpec + stats *rxStats + streams []streamState + reports chan string + ready *sync.WaitGroup +} + +func (w *rxWorker) run(done *atomic.Bool) { + pinTo(w.cpu) + + 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, _ := newMmsghdrs(bufs) + + w.ready.Done() + + for !done.Load() { + n, err := recvmmsg(w.fd, hdrs, unix.MSG_WAITFORONE) + if n <= 0 { + if err != nil && err != unix.EAGAIN && err != unix.EINTR { + w.report(fmt.Sprintf("recvmmsg: %v", err)) + } + continue + } + for i := 0; i < n; i++ { + buf := bufs[i][:int(hdrs[i].len)] + p, ok := parseHeader(buf) + if !ok { + w.stats.badMagic.Add(1) + continue + } + w.stats.frames.Add(1) + w.stats.bytes.Add(uint64(len(buf))) + + if int(p.stream) < len(w.streams) { + st := &w.streams[p.stream] + st.count.Add(1) + for { + old := st.maxSeq.Load() + if p.seq <= old || st.maxSeq.CompareAndSwap(old, p.seq) { + break + } + } + } + + if !w.verify { + continue + } + if p.payLen > len(w.spec.ref) { + w.stats.badLen.Add(1) + continue + } + pay := buf[minFrame : minFrame+p.payLen] + if crc32.Checksum(pay, crcTable) == p.crc { + continue + } + w.stats.crcErr.Add(1) + off, bits := firstDiff(pay, w.spec.ref[:p.payLen]) + w.report(fmt.Sprintf("payload mismatch stream=%d seq=%d len=%d first-diff-offset=%d bits=%d", + p.stream, p.seq, p.payLen, off, bits)) + } + } +} + +func (w *rxWorker) report(msg string) { + select { + case w.reports <- msg: + default: + } +} diff --git a/sock.go b/sock.go new file mode 100644 index 0000000..4efd100 --- /dev/null +++ b/sock.go @@ -0,0 +1,137 @@ +package main + +import ( + "fmt" + "runtime" + "unsafe" + + "golang.org/x/sys/unix" +) + +type mmsghdr struct { + hdr unix.Msghdr + len uint32 + _ [4]byte +} + +func htons(v uint16) uint16 { return v<<8 | v>>8 } + +func setBufForce(fd, forceOpt, opt, size int) error { + if err := unix.SetsockoptInt(fd, unix.SOL_SOCKET, forceOpt, size); err == nil { + return nil + } + return unix.SetsockoptInt(fd, unix.SOL_SOCKET, opt, size) +} + +func sockBufSize(fd, opt int) int { + v, err := unix.GetsockoptInt(fd, unix.SOL_SOCKET, opt) + if err != nil { + return -1 + } + return v +} + +func openTxSocket(ifindex, sndbuf int) (int, error) { + fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, 0) + if err != nil { + return -1, fmt.Errorf("socket: %w", err) + } + if err := unix.Bind(fd, &unix.SockaddrLinklayer{Ifindex: ifindex}); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("bind ifindex %d: %w", ifindex, err) + } + if err := unix.SetsockoptInt(fd, unix.SOL_PACKET, unix.PACKET_QDISC_BYPASS, 1); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("qdisc bypass: %w", err) + } + if err := setBufForce(fd, unix.SO_SNDBUFFORCE, unix.SO_SNDBUF, sndbuf); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("sndbuf: %w", err) + } + return fd, nil +} + +func openRxSocket(ifindex, rcvbuf, fanoutID, fanoutMode int) (int, error) { + proto := int(htons(etherType)) + fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, proto) + if err != nil { + return -1, fmt.Errorf("socket: %w", err) + } + if err := setBufForce(fd, unix.SO_RCVBUFFORCE, unix.SO_RCVBUF, rcvbuf); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("rcvbuf: %w", err) + } + if err := unix.Bind(fd, &unix.SockaddrLinklayer{ + Protocol: htons(etherType), + Ifindex: ifindex, + }); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("bind ifindex %d: %w", ifindex, err) + } + if err := unix.SetsockoptInt(fd, unix.SOL_PACKET, unix.PACKET_IGNORE_OUTGOING, 1); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("ignore outgoing: %w", err) + } + tv := unix.Timeval{Sec: 0, Usec: 200000} + if err := unix.SetsockoptTimeval(fd, unix.SOL_SOCKET, unix.SO_RCVTIMEO, &tv); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("rcvtimeo: %w", err) + } + if fanoutMode >= 0 { + arg := (fanoutMode << 16) | (fanoutID & 0xffff) + if err := unix.SetsockoptInt(fd, unix.SOL_PACKET, unix.PACKET_FANOUT, arg); err != nil { + unix.Close(fd) + return -1, fmt.Errorf("fanout: %w", err) + } + } + return fd, nil +} + +func sendmmsg(fd int, hdrs []mmsghdr) (int, error) { + n, _, errno := unix.Syscall6(unix.SYS_SENDMMSG, uintptr(fd), + uintptr(unsafe.Pointer(&hdrs[0])), uintptr(len(hdrs)), 0, 0, 0) + if errno != 0 { + return int(n), errno + } + return int(n), nil +} + +func recvmmsg(fd int, hdrs []mmsghdr, flags int) (int, error) { + n, _, errno := unix.Syscall6(unix.SYS_RECVMMSG, uintptr(fd), + uintptr(unsafe.Pointer(&hdrs[0])), uintptr(len(hdrs)), uintptr(flags), 0, 0) + if errno != 0 { + return int(n), errno + } + return int(n), nil +} + +func newMmsghdrs(bufs [][]byte) ([]mmsghdr, []unix.Iovec) { + hdrs := make([]mmsghdr, len(bufs)) + iovs := make([]unix.Iovec, len(bufs)) + for i := range bufs { + iovs[i].Base = &bufs[i][0] + iovs[i].Len = uint64(len(bufs[i])) + hdrs[i].hdr.Iov = &iovs[i] + hdrs[i].hdr.Iovlen = 1 + } + return hdrs, iovs +} + +func pinTo(cpu int) error { + if cpu < 0 { + return nil + } + runtime.LockOSThread() + var set unix.CPUSet + set.Zero() + set.Set(cpu) + return unix.SchedSetaffinity(0, &set) +} + +func packetDrops(fd int) uint64 { + st, err := unix.GetsockoptTpacketStats(fd, unix.SOL_PACKET, unix.PACKET_STATISTICS) + if err != nil { + return 0 + } + return uint64(st.Drops) +} diff --git a/tx.go b/tx.go new file mode 100644 index 0000000..d2cff31 --- /dev/null +++ b/tx.go @@ -0,0 +1,73 @@ +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 + cpu int + stats *txStats + startTx <-chan struct{} +} + +func (w *txWorker) run(done *atomic.Bool) { + pinTo(w.cpu) + + 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) + } + } + } +}