tweak defaults and turn on gsogro on linux by default

cert/pem.go

@@ -1,8 +1,10 @@
 package cert
 
 import (
+	"encoding/hex"
 	"encoding/pem"
 	"fmt"
+	"time"
 
 	"golang.org/x/crypto/ed25519"
 )
@@ -138,6 +140,101 @@ func MarshalSigningPrivateKeyToPEM(curve Curve, b []byte) []byte {
 	}
 }
 
+// Backward compatibility functions for older API
+func MarshalX25519PublicKey(b []byte) []byte {
+	return MarshalPublicKeyToPEM(Curve_CURVE25519, b)
+}
+
+func MarshalX25519PrivateKey(b []byte) []byte {
+	return MarshalPrivateKeyToPEM(Curve_CURVE25519, b)
+}
+
+func MarshalPublicKey(curve Curve, b []byte) []byte {
+	return MarshalPublicKeyToPEM(curve, b)
+}
+
+func MarshalPrivateKey(curve Curve, b []byte) []byte {
+	return MarshalPrivateKeyToPEM(curve, b)
+}
+
+// NebulaCertificate is a compatibility wrapper for the old API
+type NebulaCertificate struct {
+	Details   NebulaCertificateDetails
+	Signature []byte
+	cert      Certificate
+}
+
+// NebulaCertificateDetails is a compatibility wrapper for certificate details
+type NebulaCertificateDetails struct {
+	Name      string
+	NotBefore time.Time
+	NotAfter  time.Time
+	PublicKey []byte
+	IsCA      bool
+	Issuer    []byte
+	Curve     Curve
+}
+
+// UnmarshalNebulaCertificateFromPEM provides backward compatibility with the old API
+func UnmarshalNebulaCertificateFromPEM(b []byte) (*NebulaCertificate, []byte, error) {
+	c, rest, err := UnmarshalCertificateFromPEM(b)
+	if err != nil {
+		return nil, rest, err
+	}
+
+	issuerBytes, err := func() ([]byte, error) {
+		issuer := c.Issuer()
+		if issuer == "" {
+			return nil, nil
+		}
+		decoded, err := hex.DecodeString(issuer)
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode issuer fingerprint: %w", err)
+		}
+		return decoded, nil
+	}()
+	if err != nil {
+		return nil, rest, err
+	}
+
+	pubKey := c.PublicKey()
+	if pubKey != nil {
+		pubKey = append([]byte(nil), pubKey...)
+	}
+
+	sig := c.Signature()
+	if sig != nil {
+		sig = append([]byte(nil), sig...)
+	}
+
+	return &NebulaCertificate{
+		Details: NebulaCertificateDetails{
+			Name:      c.Name(),
+			NotBefore: c.NotBefore(),
+			NotAfter:  c.NotAfter(),
+			PublicKey: pubKey,
+			IsCA:      c.IsCA(),
+			Issuer:    issuerBytes,
+			Curve:     c.Curve(),
+		},
+		Signature: sig,
+		cert:      c,
+	}, rest, nil
+}
+
+// IssuerString returns the issuer in hex format for compatibility
+func (n *NebulaCertificate) IssuerString() string {
+	if n.Details.Issuer == nil {
+		return ""
+	}
+	return hex.EncodeToString(n.Details.Issuer)
+}
+
+// Certificate returns the underlying certificate (read-only)
+func (n *NebulaCertificate) Certificate() Certificate {
+	return n.cert
+}
+
 // UnmarshalPrivateKeyFromPEM will try to unmarshal the first pem block in a byte array, returning any non
 // consumed data or an error on failure
 func UnmarshalPrivateKeyFromPEM(b []byte) ([]byte, []byte, Curve, error) {

cmd/gso/gso.go

@@ -1,191 +0,0 @@
-package main
-
-import (
-	"encoding/binary"
-	"errors"
-	"flag"
-	"fmt"
-	"log"
-	"net"
-	"net/netip"
-	"time"
-	"unsafe"
-
-	"golang.org/x/sys/unix"
-)
-
-const (
-	// UDP_SEGMENT enables GSO segmentation
-	UDP_SEGMENT = 103
-	// Maximum GSO segment size (typical MTU - headers)
-	maxGSOSize = 1400
-)
-
-func main() {
-	destAddr := flag.String("dest", "10.4.0.16:4202", "Destination address")
-	gsoSize := flag.Int("gso", 1400, "GSO segment size")
-	totalSize := flag.Int("size", 14000, "Total payload size to send")
-	count := flag.Int("count", 1, "Number of packets to send")
-	flag.Parse()
-
-	if *gsoSize > maxGSOSize {
-		log.Fatalf("GSO size %d exceeds maximum %d", *gsoSize, maxGSOSize)
-	}
-
-	// Resolve destination address
-	_, err := net.ResolveUDPAddr("udp", *destAddr)
-	if err != nil {
-		log.Fatalf("Failed to resolve address: %v", err)
-	}
-
-	// Create a raw UDP socket with GSO support
-	fd, err := unix.Socket(unix.AF_INET, unix.SOCK_DGRAM, unix.IPPROTO_UDP)
-	if err != nil {
-		log.Fatalf("Failed to create socket: %v", err)
-	}
-	defer unix.Close(fd)
-
-	// Bind to a local address
-	localAddr := &unix.SockaddrInet4{
-		Port: 0, // Let the system choose a port
-	}
-	if err := unix.Bind(fd, localAddr); err != nil {
-		log.Fatalf("Failed to bind socket: %v", err)
-	}
-
-	fmt.Printf("Sending UDP packets with GSO enabled\n")
-	fmt.Printf("Destination: %s\n", *destAddr)
-	fmt.Printf("GSO segment size: %d bytes\n", *gsoSize)
-	fmt.Printf("Total payload size: %d bytes\n", *totalSize)
-	fmt.Printf("Number of packets: %d\n\n", *count)
-
-	// Create payload
-	payload := make([]byte, *totalSize)
-	for i := range payload {
-		payload[i] = byte(i % 256)
-	}
-
-	dest := netip.MustParseAddrPort(*destAddr)
-
-	//if err := unix.SetsockoptInt(fd, unix.SOL_UDP, unix.UDP_SEGMENT, 1400); err != nil {
-	//	panic(err)
-	//}
-
-	for i := 0; i < *count; i++ {
-		err := WriteBatch(fd, payload, dest, uint16(*gsoSize), true)
-		if err != nil {
-			log.Printf("Send error on packet %d: %v", i, err)
-			continue
-		}
-
-		if (i+1)%100 == 0 || i == *count-1 {
-			fmt.Printf("Sent %d packets\n", i+1)
-		}
-	}
-	fmt.Printf("now, let's send without the correct ctrl header\n")
-	time.Sleep(time.Second)
-	for i := 0; i < *count; i++ {
-		err := WriteBatch(fd, payload, dest, uint16(*gsoSize), false)
-		if err != nil {
-			log.Printf("Send error on packet %d: %v", i, err)
-			continue
-		}
-
-		if (i+1)%100 == 0 || i == *count-1 {
-			fmt.Printf("Sent %d packets\n", i+1)
-		}
-	}
-
-}
-
-func WriteBatch(fd int, payload []byte, addr netip.AddrPort, segSize uint16, withHeader bool) error {
-	msgs := make([]rawMessage, 0, 1)
-	iovs := make([]iovec, 0, 1)
-	names := make([][unix.SizeofSockaddrInet6]byte, 0, 1)
-
-	sent := 0
-
-	pkts := []BatchPacket{
-		{
-			Payload: payload,
-			Addr:    addr,
-		},
-	}
-
-	for _, pkt := range pkts {
-		if len(pkt.Payload) == 0 {
-			sent++
-			continue
-		}
-
-		msgs = append(msgs, rawMessage{})
-		iovs = append(iovs, iovec{})
-		names = append(names, [unix.SizeofSockaddrInet6]byte{})
-
-		idx := len(msgs) - 1
-		msg := &msgs[idx]
-		iov := &iovs[idx]
-		name := &names[idx]
-
-		setIovecSlice(iov, pkt.Payload)
-		msg.Hdr.Iov = iov
-		msg.Hdr.Iovlen = 1
-
-		if withHeader {
-			setRawMessageControl(msg, buildGSOControlMessage(segSize)) //
-		} else {
-			setRawMessageControl(msg, nil) //
-		}
-
-		msg.Hdr.Flags = 0
-
-		nameLen, err := encodeSockaddr(name[:], pkt.Addr)
-		if err != nil {
-			return err
-		}
-		msg.Hdr.Name = &name[0]
-		msg.Hdr.Namelen = nameLen
-	}
-
-	if len(msgs) == 0 {
-		return errors.New("nothing to write")
-	}
-
-	offset := 0
-	for offset < len(msgs) {
-		n, _, errno := unix.Syscall6(
-			unix.SYS_SENDMMSG,
-			uintptr(fd),
-			uintptr(unsafe.Pointer(&msgs[offset])),
-			uintptr(len(msgs)-offset),
-			0,
-			0,
-			0,
-		)
-
-		if errno != 0 {
-			if errno == unix.EINTR {
-				continue
-			}
-			return &net.OpError{Op: "sendmmsg", Err: errno}
-		}
-
-		if n == 0 {
-			break
-		}
-		offset += int(n)
-	}
-
-	return nil
-}
-
-func buildGSOControlMessage(segSize uint16) []byte {
-	control := make([]byte, unix.CmsgSpace(2))
-	hdr := (*unix.Cmsghdr)(unsafe.Pointer(&control[0]))
-	hdr.Level = unix.SOL_UDP
-	hdr.Type = unix.UDP_SEGMENT
-	setCmsgLen(hdr, unix.CmsgLen(2))
-	binary.NativeEndian.PutUint16(control[unix.CmsgLen(0):unix.CmsgLen(0)+2], uint16(segSize))
-
-	return control
-}

cmd/gso/helper.go

@@ -1,85 +0,0 @@
-package main
-
-import (
-	"encoding/binary"
-	"fmt"
-	"net/netip"
-	"unsafe"
-
-	"golang.org/x/sys/unix"
-)
-
-type iovec struct {
-	Base *byte
-	Len  uint64
-}
-
-type msghdr struct {
-	Name       *byte
-	Namelen    uint32
-	Pad0       [4]byte
-	Iov        *iovec
-	Iovlen     uint64
-	Control    *byte
-	Controllen uint64
-	Flags      int32
-	Pad1       [4]byte
-}
-
-type rawMessage struct {
-	Hdr  msghdr
-	Len  uint32
-	Pad0 [4]byte
-}
-
-type BatchPacket struct {
-	Payload []byte
-	Addr    netip.AddrPort
-}
-
-func encodeSockaddr(dst []byte, addr netip.AddrPort) (uint32, error) {
-	if addr.Addr().Is4() {
-		if !addr.Addr().Is4() {
-			return 0, fmt.Errorf("Listener is IPv4, but writing to IPv6 remote")
-		}
-		var sa unix.RawSockaddrInet4
-		sa.Family = unix.AF_INET
-		sa.Addr = addr.Addr().As4()
-		binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&sa.Port))[:], addr.Port())
-		size := unix.SizeofSockaddrInet4
-		copy(dst[:size], (*(*[unix.SizeofSockaddrInet4]byte)(unsafe.Pointer(&sa)))[:])
-		return uint32(size), nil
-	}
-
-	var sa unix.RawSockaddrInet6
-	sa.Family = unix.AF_INET6
-	sa.Addr = addr.Addr().As16()
-	binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&sa.Port))[:], addr.Port())
-	size := unix.SizeofSockaddrInet6
-	copy(dst[:size], (*(*[unix.SizeofSockaddrInet6]byte)(unsafe.Pointer(&sa)))[:])
-	return uint32(size), nil
-}
-
-func setRawMessageControl(msg *rawMessage, buf []byte) {
-	if len(buf) == 0 {
-		msg.Hdr.Control = nil
-		msg.Hdr.Controllen = 0
-		return
-	}
-	msg.Hdr.Control = &buf[0]
-	msg.Hdr.Controllen = uint64(len(buf))
-}
-
-func setCmsgLen(h *unix.Cmsghdr, l int) {
-	h.Len = uint64(l)
-}
-
-func setIovecSlice(iov *iovec, b []byte) {
-	if len(b) == 0 {
-		iov.Base = nil
-		iov.Len = 0
-		return
-	}
-	iov.Base = &b[0]
-	iov.Len = uint64(len(b))
-}

connection_state.go

@@ -15,7 +15,7 @@ import (
 
 // TODO: In a 5Gbps test, 1024 is not sufficient. With a 1400 MTU this is about 1.4Gbps of window, assuming full packets.
 // 4092 should be sufficient for 5Gbps
-const ReplayWindow = 8192
+const ReplayWindow = 4096
 
 type ConnectionState struct {
 	eKey           *NebulaCipherState

examples/config.yml

@@ -132,6 +132,13 @@ listen:
   # Sets the max number of packets to pull from the kernel for each syscall (under systems that support recvmmsg)
   # default is 64, does not support reload
   #batch: 64
+
+# Control batching between UDP and TUN pipelines
+#batch:
+#  inbound_size: 32        # packets to queue from UDP before handing to workers
+#  outbound_size: 32       # packets to queue from TUN before handing to workers
+#  flush_interval: 50us    # flush partially filled batches after this duration
+#  max_outstanding: 1028   # batches buffered per routine on each channel
   # Configure socket buffers for the udp side (outside), leave unset to use the system defaults. Values will be doubled by the kernel
   # Default is net.core.rmem_default and net.core.wmem_default (/proc/sys/net/core/rmem_default and /proc/sys/net/core/rmem_default)
   # Maximum is limited by memory in the system, SO_RCVBUFFORCE and SO_SNDBUFFORCE is used to avoid having to raise the system wide

inside.go

@@ -11,19 +11,19 @@ import (
 	"github.com/slackhq/nebula/routing"
 )
 
-func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet, nb, out []byte, q int, localCache firewall.ConntrackCache) {
+func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet, nb, out []byte, queue func(netip.AddrPort, int), q int, localCache firewall.ConntrackCache) bool {
 	err := newPacket(packet, false, fwPacket)
 	if err != nil {
 		if f.l.Level >= logrus.DebugLevel {
 			f.l.WithField("packet", packet).Debugf("Error while validating outbound packet: %s", err)
 		}
-		return
+		return false
 	}
 
 	// Ignore local broadcast packets
 	if f.dropLocalBroadcast {
 		if f.myBroadcastAddrsTable.Contains(fwPacket.RemoteAddr) {
-			return
+			return false
 		}
 	}
 
@@ -40,12 +40,12 @@ func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet
 		}
 		// Otherwise, drop. On linux, we should never see these packets - Linux
 		// routes packets from the nebula addr to the nebula addr through the loopback device.
-		return
+		return false
 	}
 
 	// Ignore multicast packets
 	if f.dropMulticast && fwPacket.RemoteAddr.IsMulticast() {
-		return
+		return false
 	}
 
 	hostinfo, ready := f.getOrHandshakeConsiderRouting(fwPacket, func(hh *HandshakeHostInfo) {
@@ -59,26 +59,26 @@ func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet
 				WithField("fwPacket", fwPacket).
 				Debugln("dropping outbound packet, vpnAddr not in our vpn networks or in unsafe networks")
 		}
-		return
+		return false
 	}
 
 	if !ready {
-		return
+		return false
 	}
 
 	dropReason := f.firewall.Drop(*fwPacket, false, hostinfo, f.pki.GetCAPool(), localCache)
 	if dropReason == nil {
-		f.sendNoMetrics(header.Message, 0, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, packet, nb, out, q)
-
-	} else {
-		f.rejectInside(packet, out, q)
-		if f.l.Level >= logrus.DebugLevel {
-			hostinfo.logger(f.l).
-				WithField("fwPacket", fwPacket).
-				WithField("reason", dropReason).
-				Debugln("dropping outbound packet")
-		}
+		return f.sendNoMetrics(header.Message, 0, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, packet, nb, out, queue, q)
 	}
+
+	f.rejectInside(packet, out, q)
+	if f.l.Level >= logrus.DebugLevel {
+		hostinfo.logger(f.l).
+			WithField("fwPacket", fwPacket).
+			WithField("reason", dropReason).
+			Debugln("dropping outbound packet")
+	}
+	return false
 }
 
 func (f *Interface) rejectInside(packet []byte, out []byte, q int) {
@@ -117,7 +117,7 @@ func (f *Interface) rejectOutside(packet []byte, ci *ConnectionState, hostinfo *
 		return
 	}
 
-	f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, out, nb, packet, q)
+	_ = f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, out, nb, packet, nil, q)
 }
 
 // Handshake will attempt to initiate a tunnel with the provided vpn address if it is within our vpn networks. This is a no-op if the tunnel is already established or being established
@@ -228,7 +228,7 @@ func (f *Interface) sendMessageNow(t header.MessageType, st header.MessageSubTyp
 		return
 	}
 
-	f.sendNoMetrics(header.Message, st, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, p, nb, out, 0)
+	_ = f.sendNoMetrics(header.Message, st, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, p, nb, out, nil, 0)
 }
 
 // SendMessageToVpnAddr handles real addr:port lookup and sends to the current best known address for vpnAddr
@@ -258,12 +258,12 @@ func (f *Interface) SendMessageToHostInfo(t header.MessageType, st header.Messag
 
 func (f *Interface) send(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, p, nb, out []byte) {
 	f.messageMetrics.Tx(t, st, 1)
-	f.sendNoMetrics(t, st, ci, hostinfo, netip.AddrPort{}, p, nb, out, 0)
+	_ = f.sendNoMetrics(t, st, ci, hostinfo, netip.AddrPort{}, p, nb, out, nil, 0)
 }
 
 func (f *Interface) sendTo(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte) {
 	f.messageMetrics.Tx(t, st, 1)
-	f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out, 0)
+	_ = f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out, nil, 0)
 }
 
 // SendVia sends a payload through a Relay tunnel. No authentication or encryption is done
@@ -331,9 +331,12 @@ func (f *Interface) SendVia(via *HostInfo,
 	f.connectionManager.RelayUsed(relay.LocalIndex)
 }
 
-func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte, q int) {
+// sendNoMetrics encrypts and writes/queues an outbound packet. It returns true
+// when the payload has been handed to a caller-provided queue (meaning the
+// caller is responsible for flushing it later).
+func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte, queue func(netip.AddrPort, int), q int) bool {
 	if ci.eKey == nil {
-		return
+		return false
 	}
 	useRelay := !remote.IsValid() && !hostinfo.remote.IsValid()
 	fullOut := out
@@ -380,32 +383,39 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType
 			WithField("udpAddr", remote).WithField("counter", c).
 			WithField("attemptedCounter", c).
 			Error("Failed to encrypt outgoing packet")
-		return
+		return false
 	}
 
-	if remote.IsValid() {
-		err = f.writers[q].WriteTo(out, remote)
-		if err != nil {
-			hostinfo.logger(f.l).WithError(err).
-				WithField("udpAddr", remote).Error("Failed to write outgoing packet")
-		}
-	} else if hostinfo.remote.IsValid() {
-		err = f.writers[q].WriteTo(out, hostinfo.remote)
-		if err != nil {
-			hostinfo.logger(f.l).WithError(err).
-				WithField("udpAddr", remote).Error("Failed to write outgoing packet")
-		}
-	} else {
-		// Try to send via a relay
-		for _, relayIP := range hostinfo.relayState.CopyRelayIps() {
-			relayHostInfo, relay, err := f.hostMap.QueryVpnAddrsRelayFor(hostinfo.vpnAddrs, relayIP)
-			if err != nil {
-				hostinfo.relayState.DeleteRelay(relayIP)
-				hostinfo.logger(f.l).WithField("relay", relayIP).WithError(err).Info("sendNoMetrics failed to find HostInfo")
-				continue
-			}
-			f.SendVia(relayHostInfo, relay, out, nb, fullOut[:header.Len+len(out)], true)
-			break
-		}
+	dest := remote
+	if !dest.IsValid() {
+		dest = hostinfo.remote
 	}
+
+	if dest.IsValid() {
+		if queue != nil {
+			queue(dest, len(out))
+			return true
+		}
+
+		err = f.writers[q].WriteTo(out, dest)
+		if err != nil {
+			hostinfo.logger(f.l).WithError(err).
+				WithField("udpAddr", dest).Error("Failed to write outgoing packet")
+		}
+		return false
+	}
+
+	// Try to send via a relay
+	for _, relayIP := range hostinfo.relayState.CopyRelayIps() {
+		relayHostInfo, relay, err := f.hostMap.QueryVpnAddrsRelayFor(hostinfo.vpnAddrs, relayIP)
+		if err != nil {
+			hostinfo.relayState.DeleteRelay(relayIP)
+			hostinfo.logger(f.l).WithField("relay", relayIP).WithError(err).Info("sendNoMetrics failed to find HostInfo")
+			continue
+		}
+		f.SendVia(relayHostInfo, relay, out, nb, fullOut[:header.Len+len(out)], true)
+		break
+	}
+
+	return false
 }

interface.go

@@ -22,7 +22,18 @@ import (
 	"github.com/slackhq/nebula/udp"
 )
 
-const mtu = 9001
+const (
+	mtu = 9001
+
+	inboundBatchSizeDefault      = 128
+	outboundBatchSizeDefault     = 64
+	batchFlushIntervalDefault    = 12 * time.Microsecond
+	maxOutstandingBatchesDefault = 8
+	sendBatchSizeDefault         = 64
+	maxPendingPacketsDefault     = 32
+	maxPendingBytesDefault       = 64 * 1024
+	maxSendBufPerRoutineDefault  = 16
+)
 
 type InterfaceConfig struct {
 	HostMap            *HostMap
@@ -48,9 +59,20 @@ type InterfaceConfig struct {
 	reQueryWait     time.Duration
 
 	ConntrackCacheTimeout time.Duration
+	BatchConfig           BatchConfig
 	l                     *logrus.Logger
 }
 
+type BatchConfig struct {
+	InboundBatchSize      int
+	OutboundBatchSize     int
+	FlushInterval         time.Duration
+	MaxOutstandingPerChan int
+	MaxPendingPackets     int
+	MaxPendingBytes       int
+	MaxSendBuffersPerChan int
+}
+
 type Interface struct {
 	hostMap               *HostMap
 	outside               udp.Conn
@@ -96,9 +118,157 @@ type Interface struct {
 
 	l *logrus.Logger
 
-	pktPool  *packet.Pool
-	inbound  chan *packet.Packet
-	outbound chan *packet.Packet
+	inPool  sync.Pool
+	inbound []chan *packetBatch
+
+	outPool  sync.Pool
+	outbound []chan *outboundBatch
+
+	packetBatchPool   sync.Pool
+	outboundBatchPool sync.Pool
+
+	sendPool      sync.Pool
+	sendBufCache  [][]*[]byte
+	sendBatchSize int
+
+	inboundBatchSize      int
+	outboundBatchSize     int
+	batchFlushInterval    time.Duration
+	maxOutstandingPerChan int
+	maxPendingPackets     int
+	maxPendingBytes       int
+	maxSendBufPerRoutine  int
+}
+
+type outboundSend struct {
+	buf    *[]byte
+	length int
+	addr   netip.AddrPort
+}
+
+type packetBatch struct {
+	packets []*packet.Packet
+}
+
+func newPacketBatch(capacity int) *packetBatch {
+	return &packetBatch{
+		packets: make([]*packet.Packet, 0, capacity),
+	}
+}
+
+func (b *packetBatch) add(p *packet.Packet) {
+	b.packets = append(b.packets, p)
+}
+
+func (b *packetBatch) reset() {
+	for i := range b.packets {
+		b.packets[i] = nil
+	}
+	b.packets = b.packets[:0]
+}
+
+func (f *Interface) getPacketBatch() *packetBatch {
+	if v := f.packetBatchPool.Get(); v != nil {
+		b := v.(*packetBatch)
+		b.reset()
+		return b
+	}
+	return newPacketBatch(f.inboundBatchSize)
+}
+
+func (f *Interface) releasePacketBatch(b *packetBatch) {
+	b.reset()
+	f.packetBatchPool.Put(b)
+}
+
+type outboundBatch struct {
+	payloads []*[]byte
+}
+
+func newOutboundBatch(capacity int) *outboundBatch {
+	return &outboundBatch{payloads: make([]*[]byte, 0, capacity)}
+}
+
+func (b *outboundBatch) add(buf *[]byte) {
+	b.payloads = append(b.payloads, buf)
+}
+
+func (b *outboundBatch) reset() {
+	for i := range b.payloads {
+		b.payloads[i] = nil
+	}
+	b.payloads = b.payloads[:0]
+}
+
+func (f *Interface) getOutboundBatch() *outboundBatch {
+	if v := f.outboundBatchPool.Get(); v != nil {
+		b := v.(*outboundBatch)
+		b.reset()
+		return b
+	}
+	return newOutboundBatch(f.outboundBatchSize)
+}
+
+func (f *Interface) releaseOutboundBatch(b *outboundBatch) {
+	b.reset()
+	f.outboundBatchPool.Put(b)
+}
+
+func (f *Interface) getSendBuffer(q int) *[]byte {
+	cache := f.sendBufCache[q]
+	if n := len(cache); n > 0 {
+		buf := cache[n-1]
+		f.sendBufCache[q] = cache[:n-1]
+		*buf = (*buf)[:0]
+		return buf
+	}
+	if v := f.sendPool.Get(); v != nil {
+		buf := v.(*[]byte)
+		*buf = (*buf)[:0]
+		return buf
+	}
+	b := make([]byte, mtu)
+	return &b
+}
+
+func (f *Interface) releaseSendBuffer(q int, buf *[]byte) {
+	if buf == nil {
+		return
+	}
+	*buf = (*buf)[:0]
+	cache := f.sendBufCache[q]
+	if len(cache) < f.maxSendBufPerRoutine {
+		f.sendBufCache[q] = append(cache, buf)
+		return
+	}
+	f.sendPool.Put(buf)
+}
+
+func (f *Interface) flushSendQueue(q int, pending *[]outboundSend, pendingBytes *int) {
+	if len(*pending) == 0 {
+		return
+	}
+
+	batch := make([]udp.BatchPacket, len(*pending))
+	for i, entry := range *pending {
+		batch[i] = udp.BatchPacket{
+			Payload: (*entry.buf)[:entry.length],
+			Addr:    entry.addr,
+		}
+	}
+
+	sent, err := f.writers[q].WriteBatch(batch)
+	if err != nil {
+		f.l.WithError(err).WithField("sent", sent).Error("Failed to batch send packets")
+	}
+
+	for _, entry := range *pending {
+		f.releaseSendBuffer(q, entry.buf)
+	}
+	*pending = (*pending)[:0]
+	if pendingBytes != nil {
+		*pendingBytes = 0
+	}
 }
 
 type EncWriter interface {
@@ -168,6 +338,29 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
 	}
 
 	cs := c.pki.getCertState()
+
+	bc := c.BatchConfig
+	if bc.InboundBatchSize <= 0 {
+		bc.InboundBatchSize = inboundBatchSizeDefault
+	}
+	if bc.OutboundBatchSize <= 0 {
+		bc.OutboundBatchSize = outboundBatchSizeDefault
+	}
+	if bc.FlushInterval <= 0 {
+		bc.FlushInterval = batchFlushIntervalDefault
+	}
+	if bc.MaxOutstandingPerChan <= 0 {
+		bc.MaxOutstandingPerChan = maxOutstandingBatchesDefault
+	}
+	if bc.MaxPendingPackets <= 0 {
+		bc.MaxPendingPackets = maxPendingPacketsDefault
+	}
+	if bc.MaxPendingBytes <= 0 {
+		bc.MaxPendingBytes = maxPendingBytesDefault
+	}
+	if bc.MaxSendBuffersPerChan <= 0 {
+		bc.MaxSendBuffersPerChan = maxSendBufPerRoutineDefault
+	}
 	ifce := &Interface{
 		pki:                   c.pki,
 		hostMap:               c.HostMap,
@@ -200,14 +393,48 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
 			dropped: metrics.GetOrRegisterCounter("hostinfo.cached_packets.dropped", nil),
 		},
 
-		//TODO: configurable size
-		inbound:  make(chan *packet.Packet, 2048),
-		outbound: make(chan *packet.Packet, 2048),
+		inbound:  make([]chan *packetBatch, c.routines),
+		outbound: make([]chan *outboundBatch, c.routines),
 
 		l: c.l,
+
+		inboundBatchSize:      bc.InboundBatchSize,
+		outboundBatchSize:     bc.OutboundBatchSize,
+		batchFlushInterval:    bc.FlushInterval,
+		maxOutstandingPerChan: bc.MaxOutstandingPerChan,
+		maxPendingPackets:     bc.MaxPendingPackets,
+		maxPendingBytes:       bc.MaxPendingBytes,
+		maxSendBufPerRoutine:  bc.MaxSendBuffersPerChan,
+		sendBatchSize:         bc.OutboundBatchSize,
 	}
 
-	ifce.pktPool = packet.GetPool()
+	for i := 0; i < c.routines; i++ {
+		ifce.inbound[i] = make(chan *packetBatch, ifce.maxOutstandingPerChan)
+		ifce.outbound[i] = make(chan *outboundBatch, ifce.maxOutstandingPerChan)
+	}
+
+	ifce.inPool = sync.Pool{New: func() any {
+		return packet.New()
+	}}
+
+	ifce.outPool = sync.Pool{New: func() any {
+		t := make([]byte, mtu)
+		return &t
+	}}
+
+	ifce.packetBatchPool = sync.Pool{New: func() any {
+		return newPacketBatch(ifce.inboundBatchSize)
+	}}
+
+	ifce.outboundBatchPool = sync.Pool{New: func() any {
+		return newOutboundBatch(ifce.outboundBatchSize)
+	}}
+
+	ifce.sendPool = sync.Pool{New: func() any {
+		buf := make([]byte, mtu)
+		return &buf
+	}}
+	ifce.sendBufCache = make([][]*[]byte, c.routines)
 
 	ifce.tryPromoteEvery.Store(c.tryPromoteEvery)
 	ifce.reQueryEvery.Store(c.reQueryEvery)
@@ -258,21 +485,19 @@ func (f *Interface) activate() error {
 
 func (f *Interface) run(c context.Context) (func(), error) {
 	for i := 0; i < f.routines; i++ {
-		// read packets from udp and queue to f.inbound
+		// Launch n queues to read packets from udp
 		f.wg.Add(1)
 		go f.listenOut(i)
 
-		// Launch n queues to read packets from inside tun dev and queue to f.outbound
-		//todo this never stops f.wg.Add(1)
+		// Launch n queues to read packets from tun dev
+		f.wg.Add(1)
 		go f.listenIn(f.readers[i], i)
 
-		// Launch n workers to process traffic from f.inbound and smash it onto the inside of the tun
-		f.wg.Add(1)
-		go f.workerIn(i, c)
+		// Launch n queues to read packets from tun dev
 		f.wg.Add(1)
 		go f.workerIn(i, c)
 
-		// read from f.outbound and write to UDP (outside the tun)
+		// Launch n queues to read packets from tun dev
 		f.wg.Add(1)
 		go f.workerOut(i, c)
 	}
@@ -289,7 +514,41 @@ func (f *Interface) listenOut(i int) {
 		li = f.outside
 	}
 
-	err := li.ListenOut(f.pktPool.Get, f.inbound)
+	batch := f.getPacketBatch()
+	lastFlush := time.Now()
+
+	flush := func(force bool) {
+		if len(batch.packets) == 0 {
+			if force {
+				f.releasePacketBatch(batch)
+			}
+			return
+		}
+
+		f.inbound[i] <- batch
+		batch = f.getPacketBatch()
+		lastFlush = time.Now()
+	}
+
+	err := li.ListenOut(func(fromUdpAddr netip.AddrPort, payload []byte) {
+		p := f.inPool.Get().(*packet.Packet)
+		p.Payload = p.Payload[:mtu]
+		copy(p.Payload, payload)
+		p.Payload = p.Payload[:len(payload)]
+		p.Addr = fromUdpAddr
+		batch.add(p)
+
+		if len(batch.packets) >= f.inboundBatchSize || time.Since(lastFlush) >= f.batchFlushInterval {
+			flush(false)
+		}
+	})
+
+	if len(batch.packets) > 0 {
+		f.inbound[i] <- batch
+	} else {
+		f.releasePacketBatch(batch)
+	}
+
 	if err != nil && !f.closed.Load() {
 		f.l.WithError(err).Error("Error while reading packet inbound packet, closing")
 		//TODO: Trigger Control to close
@@ -302,9 +561,26 @@ func (f *Interface) listenOut(i int) {
 func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) {
 	runtime.LockOSThread()
 
+	batch := f.getOutboundBatch()
+	lastFlush := time.Now()
+
+	flush := func(force bool) {
+		if len(batch.payloads) == 0 {
+			if force {
+				f.releaseOutboundBatch(batch)
+			}
+			return
+		}
+
+		f.outbound[i] <- batch
+		batch = f.getOutboundBatch()
+		lastFlush = time.Now()
+	}
+
 	for {
-		p := f.pktPool.Get()
-		n, err := reader.Read(p.Payload)
+		p := f.outPool.Get().(*[]byte)
+		*p = (*p)[:mtu]
+		n, err := reader.Read(*p)
 		if err != nil {
 			if !f.closed.Load() {
 				f.l.WithError(err).Error("Error while reading outbound packet, closing")
@@ -313,14 +589,18 @@ func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) {
 			break
 		}
 
-		p.Payload = (p.Payload)[:n]
-		//TODO: nonblocking channel write
-		f.outbound <- p
-		//select {
-		//case f.outbound <- p:
-		//default:
-		//	f.l.Error("Dropped packet from outbound channel")
-		//}
+		*p = (*p)[:n]
+		batch.add(p)
+
+		if len(batch.payloads) >= f.outboundBatchSize || time.Since(lastFlush) >= f.batchFlushInterval {
+			flush(false)
+		}
+	}
+
+	if len(batch.payloads) > 0 {
+		f.outbound[i] <- batch
+	} else {
+		f.releaseOutboundBatch(batch)
 	}
 
 	f.l.Debugf("overlay reader %v is done", i)
@@ -337,20 +617,13 @@ func (f *Interface) workerIn(i int, ctx context.Context) {
 
 	for {
 		select {
-		case p := <-f.inbound:
-			if p.SegSize > 0 && p.SegSize < len(p.Payload) {
-				for offset := 0; offset < len(p.Payload); offset += p.SegSize {
-					end := offset + p.SegSize
-					if end > len(p.Payload) {
-						end = len(p.Payload)
-					}
-					f.readOutsidePackets(p.Addr, nil, result2[:0], p.Payload[offset:end], h, fwPacket2, lhh, nb2, i, conntrackCache.Get(f.l))
-				}
-			} else {
+		case batch := <-f.inbound[i]:
+			for _, p := range batch.packets {
 				f.readOutsidePackets(p.Addr, nil, result2[:0], p.Payload, h, fwPacket2, lhh, nb2, i, conntrackCache.Get(f.l))
+				p.Payload = p.Payload[:mtu]
+				f.inPool.Put(p)
 			}
-
-			f.pktPool.Put(p)
+			f.releasePacketBatch(batch)
 		case <-ctx.Done():
 			f.wg.Done()
 			return
@@ -362,14 +635,52 @@ func (f *Interface) workerOut(i int, ctx context.Context) {
 	conntrackCache := firewall.NewConntrackCacheTicker(f.conntrackCacheTimeout)
 	fwPacket1 := &firewall.Packet{}
 	nb1 := make([]byte, 12, 12)
-	result1 := make([]byte, mtu)
+	pending := make([]outboundSend, 0, f.sendBatchSize)
+	pendingBytes := 0
+	maxPendingPackets := f.maxPendingPackets
+	if maxPendingPackets <= 0 {
+		maxPendingPackets = f.sendBatchSize
+	}
+	maxPendingBytes := f.maxPendingBytes
+	if maxPendingBytes <= 0 {
+		maxPendingBytes = f.sendBatchSize * mtu
+	}
 
 	for {
 		select {
-		case data := <-f.outbound:
-			f.consumeInsidePacket(data.Payload, fwPacket1, nb1, result1, i, conntrackCache.Get(f.l))
-			f.pktPool.Put(data)
+		case batch := <-f.outbound[i]:
+			for _, data := range batch.payloads {
+				sendBuf := f.getSendBuffer(i)
+				buf := (*sendBuf)[:0]
+				queue := func(addr netip.AddrPort, length int) {
+					if len(pending) >= maxPendingPackets || pendingBytes+length > maxPendingBytes {
+						f.flushSendQueue(i, &pending, &pendingBytes)
+					}
+					pending = append(pending, outboundSend{
+						buf:    sendBuf,
+						length: length,
+						addr:   addr,
+					})
+					pendingBytes += length
+					if len(pending) >= f.sendBatchSize || pendingBytes >= maxPendingBytes {
+						f.flushSendQueue(i, &pending, &pendingBytes)
+					}
+				}
+				sent := f.consumeInsidePacket(*data, fwPacket1, nb1, buf, queue, i, conntrackCache.Get(f.l))
+				if !sent {
+					f.releaseSendBuffer(i, sendBuf)
+				}
+				*data = (*data)[:mtu]
+				f.outPool.Put(data)
+			}
+			f.releaseOutboundBatch(batch)
+			if len(pending) > 0 {
+				f.flushSendQueue(i, &pending, &pendingBytes)
+			}
 		case <-ctx.Done():
+			if len(pending) > 0 {
+				f.flushSendQueue(i, &pending, &pendingBytes)
+			}
 			f.wg.Done()
 			return
 		}

main.go

@@ -164,7 +164,7 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
 
 		for i := 0; i < routines; i++ {
 			l.Infof("listening on %v", netip.AddrPortFrom(listenHost, uint16(port)))
-			udpServer, err := udp.NewListener(l, listenHost, port, routines > 1, c.GetInt("listen.batch", 64))
+			udpServer, err := udp.NewListener(l, listenHost, port, routines > 1, c.GetInt("listen.batch", 128))
 			if err != nil {
 				return nil, util.NewContextualError("Failed to open udp listener", m{"queue": i}, err)
 			}
@@ -221,6 +221,16 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
 		}
 	}
 
+	batchCfg := BatchConfig{
+		InboundBatchSize:      c.GetInt("batch.inbound_size", inboundBatchSizeDefault),
+		OutboundBatchSize:     c.GetInt("batch.outbound_size", outboundBatchSizeDefault),
+		FlushInterval:         c.GetDuration("batch.flush_interval", batchFlushIntervalDefault),
+		MaxOutstandingPerChan: c.GetInt("batch.max_outstanding", maxOutstandingBatchesDefault),
+		MaxPendingPackets:     c.GetInt("batch.max_pending_packets", 0),
+		MaxPendingBytes:       c.GetInt("batch.max_pending_bytes", 0),
+		MaxSendBuffersPerChan: c.GetInt("batch.max_send_buffers_per_routine", 0),
+	}
+
 	ifConfig := &InterfaceConfig{
 		HostMap:               hostMap,
 		Inside:                tun,
@@ -242,6 +252,7 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
 		relayManager:          NewRelayManager(ctx, l, hostMap, c),
 		punchy:                punchy,
 		ConntrackCacheTimeout: conntrackCacheTimeout,
+		BatchConfig:           batchCfg,
 		l:                     l,
 	}
 

packet/packet.go

@@ -1,45 +1,12 @@
 package packet
 
-import (
-	"net/netip"
-	"sync"
-
-	"golang.org/x/sys/unix"
-)
-
-const Size = 0xffff
+import "net/netip"
 
 type Packet struct {
 	Payload []byte
-	Control []byte
-	SegSize int
 	Addr    netip.AddrPort
 }
 
 func New() *Packet {
-	return &Packet{
-		Payload: make([]byte, Size),
-		Control: make([]byte, unix.CmsgSpace(2)),
-	}
-}
-
-type Pool struct {
-	pool sync.Pool
-}
-
-var bigPool = &Pool{
-	pool: sync.Pool{New: func() any { return New() }},
-}
-
-func GetPool() *Pool {
-	return bigPool
-}
-
-func (p *Pool) Get() *Packet {
-	return p.pool.Get().(*Packet)
-}
-
-func (p *Pool) Put(x *Packet) {
-	x.Payload = x.Payload[:Size]
-	p.pool.Put(x)
+	return &Packet{Payload: make([]byte, 9001)}
 }

service/service.go

@@ -9,13 +9,10 @@ import (
 	"math"
 	"net"
 	"net/netip"
-	"os"
 	"strings"
 	"sync"
 
-	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula"
-	"github.com/slackhq/nebula/config"
 	"github.com/slackhq/nebula/overlay"
 	"golang.org/x/sync/errgroup"
 	"gvisor.dev/gvisor/pkg/buffer"
@@ -46,15 +43,7 @@ type Service struct {
 	}
 }
 
-func New(config *config.C) (*Service, error) {
-	logger := logrus.New()
-	logger.Out = os.Stdout
-
-	control, err := nebula.Main(config, false, "custom-app", logger, overlay.NewUserDeviceFromConfig)
-	if err != nil {
-		return nil, err
-	}
-
+func New(control *nebula.Control) (*Service, error) {
 	wait, err := control.Start()
 	if err != nil {
 		return nil, err

udp/conn.go

@@ -4,24 +4,30 @@ import (
 	"net/netip"
 
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/packet"
 )
 
 const MTU = 9001
 
-type EncReader func(*packet.Packet)
-
-type PacketBufferGetter func() *packet.Packet
+type EncReader func(
+	addr netip.AddrPort,
+	payload []byte,
+)
 
 type Conn interface {
 	Rebind() error
 	LocalAddr() (netip.AddrPort, error)
-	ListenOut(pg PacketBufferGetter, pc chan *packet.Packet) error
+	ListenOut(r EncReader) error
 	WriteTo(b []byte, addr netip.AddrPort) error
+	WriteBatch(pkts []BatchPacket) (int, error)
 	ReloadConfig(c *config.C)
 	Close() error
 }
 
+type BatchPacket struct {
+	Payload []byte
+	Addr    netip.AddrPort
+}
+
 type NoopConn struct{}
 
 func (NoopConn) Rebind() error {
@@ -30,12 +36,15 @@ func (NoopConn) Rebind() error {
 func (NoopConn) LocalAddr() (netip.AddrPort, error) {
 	return netip.AddrPort{}, nil
 }
-func (NoopConn) ListenOut(_ EncReader) {
-	return
+func (NoopConn) ListenOut(_ EncReader) error {
+	return nil
 }
 func (NoopConn) WriteTo(_ []byte, _ netip.AddrPort) error {
 	return nil
 }
+func (NoopConn) WriteBatch(_ []BatchPacket) (int, error) {
+	return 0, nil
+}
 func (NoopConn) ReloadConfig(_ *config.C) {
 	return
 }

udp/udp_darwin.go

@@ -140,6 +140,17 @@ func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error {
 	}
 }
 
+func (u *StdConn) WriteBatch(pkts []BatchPacket) (int, error) {
+	sent := 0
+	for _, pkt := range pkts {
+		if err := u.WriteTo(pkt.Payload, pkt.Addr); err != nil {
+			return sent, err
+		}
+		sent++
+	}
+	return sent, nil
+}
+
 func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
 	a := u.UDPConn.LocalAddr()
 
@@ -165,7 +176,7 @@ func NewUDPStatsEmitter(udpConns []Conn) func() {
 	return func() {}
 }
 
-func (u *StdConn) ListenOut(r EncReader) {
+func (u *StdConn) ListenOut(r EncReader) error {
 	buffer := make([]byte, MTU)
 
 	for {
@@ -174,14 +185,17 @@ func (u *StdConn) ListenOut(r EncReader) {
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
 				u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
-				return
+				return err
 			}
 
 			u.l.WithError(err).Error("unexpected udp socket receive error")
+			continue
 		}
 
 		r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n])
 	}
+
+	return nil
 }
 
 func (u *StdConn) Rebind() error {

udp/udp_generic.go

@@ -42,6 +42,17 @@ func (u *GenericConn) WriteTo(b []byte, addr netip.AddrPort) error {
 	return err
 }
 
+func (u *GenericConn) WriteBatch(pkts []BatchPacket) (int, error) {
+	sent := 0
+	for _, pkt := range pkts {
+		if err := u.WriteTo(pkt.Payload, pkt.Addr); err != nil {
+			return sent, err
+		}
+		sent++
+	}
+	return sent, nil
+}
+
 func (u *GenericConn) LocalAddr() (netip.AddrPort, error) {
 	a := u.UDPConn.LocalAddr()
 

udp/udp_linux.go

@@ -5,9 +5,11 @@ package udp
 
 import (
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"net"
 	"net/netip"
+	"sync"
 	"syscall"
 	"time"
 	"unsafe"
@@ -15,20 +17,43 @@ import (
 	"github.com/rcrowley/go-metrics"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/packet"
 	"golang.org/x/sys/unix"
 )
 
 var readTimeout = unix.NsecToTimeval(int64(time.Millisecond * 500))
 
+const (
+	defaultGSOMaxSegments    = 128
+	defaultGSOFlushTimeout   = 80 * time.Microsecond
+	defaultGROReadBufferSize = MTU * defaultGSOMaxSegments
+	maxGSOBatchBytes         = 0xFFFF
+)
+
+var (
+	errGSOFallback = errors.New("udp gso fallback")
+	errGSODisabled = errors.New("udp gso disabled")
+)
+
 type StdConn struct {
-	sysFd     int
-	isV4      bool
-	l         *logrus.Logger
-	batch     int
+	sysFd int
+	isV4  bool
+	l     *logrus.Logger
+	batch int
+
 	enableGRO bool
 	enableGSO bool
-	//gso       gsoState
+
+	gsoMu           sync.Mutex
+	gsoBuf          []byte
+	gsoAddr         netip.AddrPort
+	gsoSegSize      int
+	gsoSegments     int
+	gsoMaxSegments  int
+	gsoMaxBytes     int
+	gsoFlushTimeout time.Duration
+	gsoTimer        *time.Timer
+
+	groBufSize int
 }
 
 func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
@@ -73,7 +98,16 @@ func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch in
 		return nil, fmt.Errorf("unable to bind to socket: %s", err)
 	}
 
-	return &StdConn{sysFd: fd, isV4: ip.Is4(), l: l, batch: batch}, err
+	return &StdConn{
+		sysFd:           fd,
+		isV4:            ip.Is4(),
+		l:               l,
+		batch:           batch,
+		gsoMaxSegments:  defaultGSOMaxSegments,
+		gsoMaxBytes:     MTU * defaultGSOMaxSegments,
+		gsoFlushTimeout: defaultGSOFlushTimeout,
+		groBufSize:      MTU,
+	}, err
 }
 
 func (u *StdConn) Rebind() error {
@@ -122,71 +156,129 @@ func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
 	}
 }
 
-func (u *StdConn) ListenOut(pg PacketBufferGetter, pc chan *packet.Packet) error {
-	var ip netip.Addr
+func (u *StdConn) ListenOut(r EncReader) error {
+	var (
+		ip       netip.Addr
+		controls [][]byte
+	)
 
-	msgs, packets, names := u.PrepareRawMessages(u.batch, pg)
+	bufSize := u.readBufferSize()
+	msgs, buffers, names := u.PrepareRawMessages(u.batch, bufSize)
 	read := u.ReadMulti
 	if u.batch == 1 {
 		read = u.ReadSingle
 	}
 
 	for {
+		desired := u.readBufferSize()
+		if len(buffers) == 0 || cap(buffers[0]) < desired {
+			msgs, buffers, names = u.PrepareRawMessages(u.batch, desired)
+			controls = nil
+		}
+
+		if u.enableGRO {
+			if controls == nil {
+				controls = make([][]byte, len(msgs))
+				for i := range controls {
+					controls[i] = make([]byte, unix.CmsgSpace(4))
+				}
+			}
+			for i := range msgs {
+				setRawMessageControl(&msgs[i], controls[i])
+			}
+		} else if controls != nil {
+			for i := range msgs {
+				setRawMessageControl(&msgs[i], nil)
+			}
+			controls = nil
+		}
+
 		n, err := read(msgs)
 		if err != nil {
 			return err
 		}
 
 		for i := 0; i < n; i++ {
-			out := packets[i]
-			out.Payload = out.Payload[:msgs[i].Len]
-
 			// Its ok to skip the ok check here, the slicing is the only error that can occur and it will panic
 			if u.isV4 {
 				ip, _ = netip.AddrFromSlice(names[i][4:8])
 			} else {
 				ip, _ = netip.AddrFromSlice(names[i][8:24])
 			}
-			out.Addr = netip.AddrPortFrom(ip.Unmap(), binary.BigEndian.Uint16(names[i][2:4]))
-			ctrlLen := getRawMessageControlLen(&msgs[i])
-			if ctrlLen > 0 {
-				packets[i].SegSize = parseGROControl(packets[i].Control[:ctrlLen])
-			} else {
-				packets[i].SegSize = 0
+			addr := netip.AddrPortFrom(ip.Unmap(), binary.BigEndian.Uint16(names[i][2:4]))
+			payload := buffers[i][:msgs[i].Len]
+
+			if u.enableGRO && u.l.IsLevelEnabled(logrus.DebugLevel) {
+				ctrlLen := getRawMessageControlLen(&msgs[i])
+				msgFlags := getRawMessageFlags(&msgs[i])
+				u.l.WithFields(logrus.Fields{
+					"tag":         "gro-debug",
+					"stage":       "recv",
+					"payload_len": len(payload),
+					"ctrl_len":    ctrlLen,
+					"msg_flags":   msgFlags,
+				}).Debug("gro batch data")
+				if controls != nil && ctrlLen > 0 {
+					maxDump := ctrlLen
+					if maxDump > 16 {
+						maxDump = 16
+					}
+					u.l.WithFields(logrus.Fields{
+						"tag":         "gro-debug",
+						"stage":       "control-bytes",
+						"control_hex": fmt.Sprintf("%x", controls[i][:maxDump]),
+						"datalen":     ctrlLen,
+					}).Debug("gro control dump")
+				}
 			}
 
-			pc <- out
-
-			//rotate this packet out so we don't overwrite it
-			packets[i] = pg()
-			msgs[i].Hdr.Iov.Base = &packets[i].Payload[0]
-			if u.enableGRO {
-				msgs[i].Hdr.Control = &packets[i].Control[0]
-				msgs[i].Hdr.Controllen = uint64(cap(packets[i].Control))
+			sawControl := false
+			if controls != nil {
+				if ctrlLen := getRawMessageControlLen(&msgs[i]); ctrlLen > 0 {
+					if segSize, segCount := parseGROControl(controls[i][:ctrlLen]); segSize > 0 {
+						sawControl = true
+						if u.l.IsLevelEnabled(logrus.DebugLevel) {
+							u.l.WithFields(logrus.Fields{
+								"tag":        "gro-debug",
+								"stage":      "control",
+								"seg_size":   segSize,
+								"seg_count":  segCount,
+								"payloadLen": len(payload),
+							}).Debug("gro control parsed")
+						}
+						segSize = normalizeGROSegSize(segSize, segCount, len(payload))
+						if segSize > 0 && segSize < len(payload) {
+							if u.emitGROSegments(r, addr, payload, segSize) {
+								continue
+							}
+						}
+					}
+				}
 			}
 
+			if u.enableGRO && len(payload) > MTU {
+				if !sawControl && u.l.IsLevelEnabled(logrus.DebugLevel) {
+					u.l.WithFields(logrus.Fields{
+						"tag":         "gro-debug",
+						"stage":       "fallback",
+						"payload_len": len(payload),
+					}).Debug("gro control missing; splitting payload by MTU")
+				}
+				if u.emitGROSegments(r, addr, payload, MTU) {
+					continue
+				}
+			}
+
+			r(addr, payload)
 		}
 	}
 }
 
-func parseGROControl(control []byte) int {
-	if len(control) == 0 {
-		return 0
+func (u *StdConn) readBufferSize() int {
+	if u.enableGRO && u.groBufSize > MTU {
+		return u.groBufSize
 	}
-
-	cmsgs, err := unix.ParseSocketControlMessage(control)
-	if err != nil {
-		return 0
-	}
-
-	for _, c := range cmsgs {
-		if c.Header.Level == unix.SOL_UDP && c.Header.Type == unix.UDP_GRO && len(c.Data) >= 2 {
-			segSize := int(binary.LittleEndian.Uint16(c.Data[:2]))
-			return segSize
-		}
-	}
-
-	return 0
+	return MTU
 }
 
 func (u *StdConn) ReadSingle(msgs []rawMessage) (int, error) {
@@ -237,12 +329,132 @@ func (u *StdConn) ReadMulti(msgs []rawMessage) (int, error) {
 }
 
 func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort) error {
+	if u.enableGSO && ip.IsValid() {
+		if err := u.queueGSOPacket(b, ip); err == nil {
+			return nil
+		} else if !errors.Is(err, errGSOFallback) {
+			return err
+		}
+	}
+
 	if u.isV4 {
 		return u.writeTo4(b, ip)
 	}
 	return u.writeTo6(b, ip)
 }
 
+func (u *StdConn) WriteBatch(pkts []BatchPacket) (int, error) {
+	if len(pkts) == 0 {
+		return 0, nil
+	}
+
+	msgs := make([]rawMessage, 0, len(pkts))
+	iovs := make([]iovec, 0, len(pkts))
+	names := make([][unix.SizeofSockaddrInet6]byte, 0, len(pkts))
+
+	sent := 0
+
+	for _, pkt := range pkts {
+		if len(pkt.Payload) == 0 {
+			sent++
+			continue
+		}
+
+		if u.enableGSO && pkt.Addr.IsValid() {
+			if err := u.queueGSOPacket(pkt.Payload, pkt.Addr); err == nil {
+				sent++
+				continue
+			} else if !errors.Is(err, errGSOFallback) {
+				return sent, err
+			}
+		}
+
+		if !pkt.Addr.IsValid() {
+			if err := u.WriteTo(pkt.Payload, pkt.Addr); err != nil {
+				return sent, err
+			}
+			sent++
+			continue
+		}
+
+		msgs = append(msgs, rawMessage{})
+		iovs = append(iovs, iovec{})
+		names = append(names, [unix.SizeofSockaddrInet6]byte{})
+
+		idx := len(msgs) - 1
+		msg := &msgs[idx]
+		iov := &iovs[idx]
+		name := &names[idx]
+
+		setIovecSlice(iov, pkt.Payload)
+		msg.Hdr.Iov = iov
+		msg.Hdr.Iovlen = 1
+		setRawMessageControl(msg, nil)
+		msg.Hdr.Flags = 0
+
+		nameLen, err := u.encodeSockaddr(name[:], pkt.Addr)
+		if err != nil {
+			return sent, err
+		}
+		msg.Hdr.Name = &name[0]
+		msg.Hdr.Namelen = nameLen
+	}
+
+	if len(msgs) == 0 {
+		return sent, nil
+	}
+
+	offset := 0
+	for offset < len(msgs) {
+		n, _, errno := unix.Syscall6(
+			unix.SYS_SENDMMSG,
+			uintptr(u.sysFd),
+			uintptr(unsafe.Pointer(&msgs[offset])),
+			uintptr(len(msgs)-offset),
+			0,
+			0,
+			0,
+		)
+
+		if errno != 0 {
+			if errno == unix.EINTR {
+				continue
+			}
+			return sent + offset, &net.OpError{Op: "sendmmsg", Err: errno}
+		}
+
+		if n == 0 {
+			break
+		}
+		offset += int(n)
+	}
+
+	return sent + len(msgs), nil
+}
+
+func (u *StdConn) encodeSockaddr(dst []byte, addr netip.AddrPort) (uint32, error) {
+	if u.isV4 {
+		if !addr.Addr().Is4() {
+			return 0, fmt.Errorf("Listener is IPv4, but writing to IPv6 remote")
+		}
+		var sa unix.RawSockaddrInet4
+		sa.Family = unix.AF_INET
+		sa.Addr = addr.Addr().As4()
+		binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&sa.Port))[:], addr.Port())
+		size := unix.SizeofSockaddrInet4
+		copy(dst[:size], (*(*[unix.SizeofSockaddrInet4]byte)(unsafe.Pointer(&sa)))[:])
+		return uint32(size), nil
+	}
+
+	var sa unix.RawSockaddrInet6
+	sa.Family = unix.AF_INET6
+	sa.Addr = addr.Addr().As16()
+	binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&sa.Port))[:], addr.Port())
+	size := unix.SizeofSockaddrInet6
+	copy(dst[:size], (*(*[unix.SizeofSockaddrInet6]byte)(unsafe.Pointer(&sa)))[:])
+	return uint32(size), nil
+}
+
 func (u *StdConn) writeTo6(b []byte, ip netip.AddrPort) error {
 	var rsa unix.RawSockaddrInet6
 	rsa.Family = unix.AF_INET6
@@ -343,11 +555,23 @@ func (u *StdConn) ReloadConfig(c *config.C) {
 			u.l.WithError(err).Error("Failed to set listen.so_mark")
 		}
 	}
-	u.configureGRO(true)
+
+	u.configureGRO(c)
+	u.configureGSO(c)
 }
 
-func (u *StdConn) configureGRO(enable bool) {
+func (u *StdConn) configureGRO(c *config.C) {
+	if c == nil {
+		return
+	}
+
+	enable := c.GetBool("listen.enable_gro", true)
 	if enable == u.enableGRO {
+		if enable {
+			if size := c.GetInt("listen.gro_read_buffer", 0); size > 0 {
+				u.setGROBufferSize(size)
+			}
+		}
 		return
 	}
 
@@ -357,7 +581,8 @@ func (u *StdConn) configureGRO(enable bool) {
 			return
 		}
 		u.enableGRO = true
-		u.l.Info("UDP GRO enabled")
+		u.setGROBufferSize(c.GetInt("listen.gro_read_buffer", defaultGROReadBufferSize))
+		u.l.WithField("buffer_size", u.groBufSize).Info("UDP GRO enabled")
 		return
 	}
 
@@ -365,6 +590,59 @@ func (u *StdConn) configureGRO(enable bool) {
 		u.l.WithError(err).Warn("Failed to disable UDP GRO")
 	}
 	u.enableGRO = false
+	u.groBufSize = MTU
+}
+
+func (u *StdConn) configureGSO(c *config.C) {
+	enable := c.GetBool("listen.enable_gso", true)
+	if !enable {
+		u.disableGSO()
+	} else {
+		u.enableGSO = true
+	}
+
+	segments := c.GetInt("listen.gso_max_segments", defaultGSOMaxSegments)
+	if segments < 1 {
+		segments = 1
+	}
+	u.gsoMaxSegments = segments
+
+	maxBytes := c.GetInt("listen.gso_max_bytes", 0)
+	if maxBytes <= 0 {
+		maxBytes = MTU * segments
+	}
+	if maxBytes > maxGSOBatchBytes {
+		u.l.WithField("requested", maxBytes).Warn("listen.gso_max_bytes larger than UDP limit; clamping")
+		maxBytes = maxGSOBatchBytes
+	}
+	u.gsoMaxBytes = maxBytes
+
+	timeout := c.GetDuration("listen.gso_flush_timeout", defaultGSOFlushTimeout)
+	if timeout < 0 {
+		timeout = 0
+	}
+	u.gsoFlushTimeout = timeout
+}
+
+func (u *StdConn) setGROBufferSize(size int) {
+	if size < MTU {
+		size = defaultGROReadBufferSize
+	}
+	if size > maxGSOBatchBytes {
+		size = maxGSOBatchBytes
+	}
+	u.groBufSize = size
+}
+
+func (u *StdConn) disableGSO() {
+	u.gsoMu.Lock()
+	defer u.gsoMu.Unlock()
+	u.enableGSO = false
+	_ = u.flushGSOlocked()
+	u.gsoBuf = nil
+	u.gsoSegments = 0
+	u.gsoSegSize = 0
+	u.stopGSOTimerLocked()
 }
 
 func (u *StdConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error {
@@ -376,7 +654,239 @@ func (u *StdConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error {
 	return nil
 }
 
+func (u *StdConn) queueGSOPacket(b []byte, addr netip.AddrPort) error {
+	if len(b) == 0 {
+		return nil
+	}
+
+	u.gsoMu.Lock()
+	defer u.gsoMu.Unlock()
+
+	if !u.enableGSO || !addr.IsValid() || len(b) > u.gsoMaxBytes {
+		if err := u.flushGSOlocked(); err != nil {
+			return err
+		}
+		return errGSOFallback
+	}
+
+	if u.gsoSegments == 0 {
+		if cap(u.gsoBuf) < u.gsoMaxBytes {
+			u.gsoBuf = make([]byte, 0, u.gsoMaxBytes)
+		}
+		u.gsoAddr = addr
+		u.gsoSegSize = len(b)
+	} else if addr != u.gsoAddr || len(b) != u.gsoSegSize {
+		if err := u.flushGSOlocked(); err != nil {
+			return err
+		}
+		if cap(u.gsoBuf) < u.gsoMaxBytes {
+			u.gsoBuf = make([]byte, 0, u.gsoMaxBytes)
+		}
+		u.gsoAddr = addr
+		u.gsoSegSize = len(b)
+	}
+
+	if len(u.gsoBuf)+len(b) > u.gsoMaxBytes {
+		if err := u.flushGSOlocked(); err != nil {
+			return err
+		}
+		if cap(u.gsoBuf) < u.gsoMaxBytes {
+			u.gsoBuf = make([]byte, 0, u.gsoMaxBytes)
+		}
+		u.gsoAddr = addr
+		u.gsoSegSize = len(b)
+	}
+
+	u.gsoBuf = append(u.gsoBuf, b...)
+	u.gsoSegments++
+
+	if u.gsoSegments >= u.gsoMaxSegments || u.gsoFlushTimeout <= 0 {
+		return u.flushGSOlocked()
+	}
+
+	u.scheduleGSOFlushLocked()
+	return nil
+}
+
+func (u *StdConn) flushGSOlocked() error {
+	if u.gsoSegments == 0 {
+		u.stopGSOTimerLocked()
+		return nil
+	}
+
+	payload := append([]byte(nil), u.gsoBuf...)
+	addr := u.gsoAddr
+	segSize := u.gsoSegSize
+
+	u.gsoBuf = u.gsoBuf[:0]
+	u.gsoSegments = 0
+	u.gsoSegSize = 0
+	u.stopGSOTimerLocked()
+
+	if segSize <= 0 {
+		return errGSOFallback
+	}
+
+	err := u.sendSegmented(payload, addr, segSize)
+	if errors.Is(err, errGSODisabled) {
+		u.l.WithField("addr", addr).Warn("UDP GSO disabled by kernel, falling back to sendto")
+		u.enableGSO = false
+		return u.sendSegmentsIndividually(payload, addr, segSize)
+	}
+
+	return err
+}
+
+func (u *StdConn) sendSegmented(payload []byte, addr netip.AddrPort, segSize int) error {
+	if len(payload) == 0 {
+		return nil
+	}
+
+	control := make([]byte, unix.CmsgSpace(2))
+	hdr := (*unix.Cmsghdr)(unsafe.Pointer(&control[0]))
+	hdr.Level = unix.SOL_UDP
+	hdr.Type = unix.UDP_SEGMENT
+	setCmsgLen(hdr, unix.CmsgLen(2))
+	binary.NativeEndian.PutUint16(control[unix.CmsgLen(0):unix.CmsgLen(0)+2], uint16(segSize))
+
+	var sa unix.Sockaddr
+	if addr.Addr().Is4() {
+		var sa4 unix.SockaddrInet4
+		sa4.Port = int(addr.Port())
+		sa4.Addr = addr.Addr().As4()
+		sa = &sa4
+	} else {
+		var sa6 unix.SockaddrInet6
+		sa6.Port = int(addr.Port())
+		sa6.Addr = addr.Addr().As16()
+		sa = &sa6
+	}
+
+	if _, err := unix.SendmsgN(u.sysFd, payload, control, sa, 0); err != nil {
+		if errno, ok := err.(syscall.Errno); ok && (errno == unix.EINVAL || errno == unix.ENOTSUP || errno == unix.EOPNOTSUPP) {
+			return errGSODisabled
+		}
+		return &net.OpError{Op: "sendmsg", Err: err}
+	}
+	return nil
+}
+
+func (u *StdConn) sendSegmentsIndividually(buf []byte, addr netip.AddrPort, segSize int) error {
+	if segSize <= 0 {
+		return errGSOFallback
+	}
+
+	for offset := 0; offset < len(buf); offset += segSize {
+		end := offset + segSize
+		if end > len(buf) {
+			end = len(buf)
+		}
+		var err error
+		if u.isV4 {
+			err = u.writeTo4(buf[offset:end], addr)
+		} else {
+			err = u.writeTo6(buf[offset:end], addr)
+		}
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (u *StdConn) scheduleGSOFlushLocked() {
+	if u.gsoTimer == nil {
+		u.gsoTimer = time.AfterFunc(u.gsoFlushTimeout, u.gsoFlushTimer)
+		return
+	}
+	u.gsoTimer.Reset(u.gsoFlushTimeout)
+}
+
+func (u *StdConn) stopGSOTimerLocked() {
+	if u.gsoTimer != nil {
+		u.gsoTimer.Stop()
+		u.gsoTimer = nil
+	}
+}
+
+func (u *StdConn) gsoFlushTimer() {
+	u.gsoMu.Lock()
+	defer u.gsoMu.Unlock()
+	_ = u.flushGSOlocked()
+}
+
+func parseGROControl(control []byte) (int, int) {
+	if len(control) == 0 {
+		return 0, 0
+	}
+
+	cmsgs, err := unix.ParseSocketControlMessage(control)
+	if err != nil {
+		return 0, 0
+	}
+
+	for _, c := range cmsgs {
+		if c.Header.Level == unix.SOL_UDP && c.Header.Type == unix.UDP_GRO && len(c.Data) >= 2 {
+			segSize := int(binary.NativeEndian.Uint16(c.Data[:2]))
+			segCount := 0
+			if len(c.Data) >= 4 {
+				segCount = int(binary.NativeEndian.Uint16(c.Data[2:4]))
+			}
+			return segSize, segCount
+		}
+	}
+
+	return 0, 0
+}
+
+func (u *StdConn) emitGROSegments(r EncReader, addr netip.AddrPort, payload []byte, segSize int) bool {
+	if segSize <= 0 {
+		return false
+	}
+
+	for offset := 0; offset < len(payload); offset += segSize {
+		end := offset + segSize
+		if end > len(payload) {
+			end = len(payload)
+		}
+		segment := make([]byte, end-offset)
+		copy(segment, payload[offset:end])
+		r(addr, segment)
+	}
+	return true
+}
+
+func normalizeGROSegSize(segSize, segCount, total int) int {
+	if segSize <= 0 || total <= 0 {
+		return segSize
+	}
+
+	if segSize > total && segCount > 0 {
+		segSize = total / segCount
+		if segSize == 0 {
+			segSize = total
+		}
+	}
+
+	if segCount <= 1 && segSize > 0 && total > segSize {
+		calculated := total / segSize
+		if calculated <= 1 {
+			calculated = (total + segSize - 1) / segSize
+		}
+		if calculated > 1 {
+			segCount = calculated
+		}
+	}
+
+	if segSize > MTU {
+		return MTU
+	}
+
+	return segSize
+}
+
 func (u *StdConn) Close() error {
+	u.disableGSO()
 	return syscall.Close(u.sysFd)
 }
 

udp/udp_linux_32.go

@@ -30,13 +30,16 @@ type rawMessage struct {
 	Len uint32
 }
 
-func (u *StdConn) PrepareRawMessages(n int) ([]rawMessage, [][]byte, [][]byte) {
+func (u *StdConn) PrepareRawMessages(n int, bufSize int) ([]rawMessage, [][]byte, [][]byte) {
+	if bufSize <= 0 {
+		bufSize = MTU
+	}
 	msgs := make([]rawMessage, n)
 	buffers := make([][]byte, n)
 	names := make([][]byte, n)
 
 	for i := range msgs {
-		buffers[i] = make([]byte, MTU)
+		buffers[i] = make([]byte, bufSize)
 		names[i] = make([]byte, unix.SizeofSockaddrInet6)
 
 		vs := []iovec{
@@ -52,3 +55,35 @@ func (u *StdConn) PrepareRawMessages(n int) ([]rawMessage, [][]byte, [][]byte) {
 
 	return msgs, buffers, names
 }
+
+func setRawMessageControl(msg *rawMessage, buf []byte) {
+	if len(buf) == 0 {
+		msg.Hdr.Control = nil
+		msg.Hdr.Controllen = 0
+		return
+	}
+	msg.Hdr.Control = &buf[0]
+	msg.Hdr.Controllen = uint32(len(buf))
+}
+
+func getRawMessageControlLen(msg *rawMessage) int {
+	return int(msg.Hdr.Controllen)
+}
+
+func getRawMessageFlags(msg *rawMessage) int {
+	return int(msg.Hdr.Flags)
+}
+
+func setCmsgLen(h *unix.Cmsghdr, l int) {
+	h.Len = uint32(l)
+}
+
+func setIovecSlice(iov *iovec, b []byte) {
+	if len(b) == 0 {
+		iov.Base = nil
+		iov.Len = 0
+		return
+	}
+	iov.Base = &b[0]
+	iov.Len = uint32(len(b))
+}

udp/udp_linux_64.go

@@ -7,7 +7,6 @@
 package udp
 
 import (
-	"github.com/slackhq/nebula/packet"
 	"golang.org/x/sys/unix"
 )
 
@@ -34,6 +33,32 @@ type rawMessage struct {
 	Pad0 [4]byte
 }
 
+func (u *StdConn) PrepareRawMessages(n int, bufSize int) ([]rawMessage, [][]byte, [][]byte) {
+	if bufSize <= 0 {
+		bufSize = MTU
+	}
+	msgs := make([]rawMessage, n)
+	buffers := make([][]byte, n)
+	names := make([][]byte, n)
+
+	for i := range msgs {
+		buffers[i] = make([]byte, bufSize)
+		names[i] = make([]byte, unix.SizeofSockaddrInet6)
+
+		vs := []iovec{
+			{Base: &buffers[i][0], Len: uint64(len(buffers[i]))},
+		}
+
+		msgs[i].Hdr.Iov = &vs[0]
+		msgs[i].Hdr.Iovlen = uint64(len(vs))
+
+		msgs[i].Hdr.Name = &names[i][0]
+		msgs[i].Hdr.Namelen = uint32(len(names[i]))
+	}
+
+	return msgs, buffers, names
+}
+
 func setRawMessageControl(msg *rawMessage, buf []byte) {
 	if len(buf) == 0 {
 		msg.Hdr.Control = nil
@@ -48,40 +73,20 @@ func getRawMessageControlLen(msg *rawMessage) int {
 	return int(msg.Hdr.Controllen)
 }
 
+func getRawMessageFlags(msg *rawMessage) int {
+	return int(msg.Hdr.Flags)
+}
+
 func setCmsgLen(h *unix.Cmsghdr, l int) {
 	h.Len = uint64(l)
 }
 
-func (u *StdConn) PrepareRawMessages(n int, pg PacketBufferGetter) ([]rawMessage, []*packet.Packet, [][]byte) {
-	msgs := make([]rawMessage, n)
-	names := make([][]byte, n)
-
-	packets := make([]*packet.Packet, n)
-	for i := range packets {
-		packets[i] = pg()
+func setIovecSlice(iov *iovec, b []byte) {
+	if len(b) == 0 {
+		iov.Base = nil
+		iov.Len = 0
+		return
 	}
-	//todo?
-
-	for i := range msgs {
-		names[i] = make([]byte, unix.SizeofSockaddrInet6)
-
-		vs := []iovec{
-			{Base: &packets[i].Payload[0], Len: uint64(packet.Size)},
-		}
-
-		msgs[i].Hdr.Iov = &vs[0]
-		msgs[i].Hdr.Iovlen = uint64(len(vs))
-
-		msgs[i].Hdr.Name = &names[i][0]
-		msgs[i].Hdr.Namelen = uint32(len(names[i]))
-		if u.enableGRO {
-			msgs[i].Hdr.Control = &packets[i].Control[0]
-			msgs[i].Hdr.Controllen = uint64(len(packets[i].Control))
-		} else {
-			msgs[i].Hdr.Control = nil
-			msgs[i].Hdr.Controllen = 0
-		}
-	}
-
-	return msgs, packets, names
+	iov.Base = &b[0]
+	iov.Len = uint64(len(b))
 }

udp/udp_rio_windows.go

@@ -304,6 +304,17 @@ func (u *RIOConn) WriteTo(buf []byte, ip netip.AddrPort) error {
 	return winrio.SendEx(u.rq, dataBuffer, 1, nil, addressBuffer, nil, nil, 0, 0)
 }
 
+func (u *RIOConn) WriteBatch(pkts []BatchPacket) (int, error) {
+	sent := 0
+	for _, pkt := range pkts {
+		if err := u.WriteTo(pkt.Payload, pkt.Addr); err != nil {
+			return sent, err
+		}
+		sent++
+	}
+	return sent, nil
+}
+
 func (u *RIOConn) LocalAddr() (netip.AddrPort, error) {
 	sa, err := windows.Getsockname(u.sock)
 	if err != nil {

udp/udp_tester.go

@@ -106,6 +106,17 @@ func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error {
 	return nil
 }
 
+func (u *TesterConn) WriteBatch(pkts []BatchPacket) (int, error) {
+	sent := 0
+	for _, pkt := range pkts {
+		if err := u.WriteTo(pkt.Payload, pkt.Addr); err != nil {
+			return sent, err
+		}
+		sent++
+	}
+	return sent, nil
+}
+
 func (u *TesterConn) ListenOut(r EncReader) {
 	for {
 		p, ok := <-u.RxPackets