drop runtime.LockOSThread() on listenOut

CODEOWNERS

@@ -1 +0,0 @@
-#ECCN:Open Source

dist/wireshark/nebula.lua

@@ -84,24 +84,30 @@ end
 
 function nebula.prefs_changed()
     if default_settings.all_ports == nebula.prefs.all_ports and default_settings.port == nebula.prefs.port then
+        -- Nothing changed, bail
         return
     end
 
-    -- Remove all existing registrations
+    -- Remove our old dissector
     DissectorTable.get("udp.port"):remove_all(nebula)
 
-    if nebula.prefs.all_ports then
+    if nebula.prefs.all_ports and default_settings.all_ports ~= nebula.prefs.all_ports then
-        -- Register on every port for hole punch capture
+        default_settings.all_port = nebula.prefs.all_ports
+
         for i=0, 65535 do
             DissectorTable.get("udp.port"):add(i, nebula)
         end
-    else
+
-        -- Register on the configured port only
+        -- no need to establish again on specific ports
-        DissectorTable.get("udp.port"):add(nebula.prefs.port, nebula)
+        return
     end
 
-    default_settings.all_ports = nebula.prefs.all_ports
+
-    default_settings.port = nebula.prefs.port
+    if default_settings.all_ports ~= nebula.prefs.all_ports then
+        -- Add our new port dissector
+        default_settings.port = nebula.prefs.port
+        DissectorTable.get("udp.port"):add(default_settings.port, nebula)
+    end
 end
 
 DissectorTable.get("udp.port"):add(default_settings.port, nebula)

e2e/tunnels_test.go

@@ -12,8 +12,6 @@ import (
 	"github.com/slackhq/nebula/cert"
 	"github.com/slackhq/nebula/cert_test"
 	"github.com/slackhq/nebula/e2e/router"
-	"github.com/slackhq/nebula/header"
-	"github.com/slackhq/nebula/udp"
 	"github.com/stretchr/testify/assert"
 	"gopkg.in/yaml.v3"
 )
@@ -367,106 +365,3 @@ func TestCrossStackRelaysWork(t *testing.T) {
 	//theirControl.Stop()
 	//relayControl.Stop()
 }
-
-func TestCloseTunnelAuthenticated(t *testing.T) {
-	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
-	myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", m{"tunnels": m{"drop_inactive": true, "inactivity_timeout": "5s"}})
-	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", m{"tunnels": m{"drop_inactive": true, "inactivity_timeout": "10m"}})
-
-	// Share our underlay information
-	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
-	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
-
-	// Start the servers
-	myControl.Start()
-	theirControl.Start()
-
-	r := router.NewR(t, myControl, theirControl)
-
-	r.Log("Assert the tunnel between me and them works")
-	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
-
-	r.Log("Close the tunnel")
-	myControl.CloseTunnel(theirVpnIpNet[0].Addr(), false)
-	r.FlushAll()
-
-	waitStart := time.Now()
-	for {
-		myIndexes := len(myControl.GetHostmap().Indexes)
-		theirIndexes := len(theirControl.GetHostmap().Indexes)
-		if myIndexes == 0 && theirIndexes == 0 {
-			break
-		}
-
-		since := time.Since(waitStart)
-		r.Logf("my tunnels: %v; their tunnels: %v; duration: %v", myIndexes, theirIndexes, since)
-		if since > time.Second*6 {
-			t.Fatal("Tunnel should have been declared inactive after 2 seconds and before 6 seconds")
-		}
-
-		time.Sleep(1 * time.Second)
-		//r.FlushAll()
-	}
-
-	r.Logf("Happy path success, tunnels were dropped within %v", time.Since(waitStart))
-
-	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
-	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
-	r.Log("Assert another tunnel between me and them works")
-	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
-	hi := myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false)
-	if hi == nil {
-		t.Fatal("There is no hostinfo for this tunnel")
-	}
-	myHi := theirControl.GetHostInfoByVpnAddr(myVpnIpNet[0].Addr(), false)
-	if myHi == nil {
-		t.Fatal("There is no hostinfo for my tunnel")
-	}
-	r.Log("It does")
-
-	buf := make([]byte, 1024)
-	hdr := header.H{
-		Version:        1,
-		Type:           header.CloseTunnel,
-		Subtype:        0,
-		Reserved:       0,
-		RemoteIndex:    hi.RemoteIndex,
-		MessageCounter: 5,
-	}
-	out, err := hdr.Encode(buf)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	pkt := &udp.Packet{
-		To:   hi.CurrentRemote,
-		From: myHi.CurrentRemote,
-		Data: out,
-	}
-	r.InjectUDPPacket(myControl, theirControl, pkt)
-	r.Log("Injected bogus close tunnel. Let's see!")
-	waitStart = time.Now()
-	for {
-		myIndexes := len(myControl.GetHostmap().Indexes)
-		theirIndexes := len(theirControl.GetHostmap().Indexes)
-		if myIndexes == 0 {
-			t.Fatal("myIndexes should not be 0")
-		}
-		if theirIndexes == 0 {
-			t.Fatal("theirIndexes should not be 0, they should have rejected this bogus packet")
-		}
-
-		since := time.Since(waitStart)
-		r.Logf("my tunnels: %v; their tunnels: %v; duration: %v", myIndexes, theirIndexes, since)
-		if since > time.Second*4 {
-			t.Log("The tunnel would have been gone by now")
-			break
-		}
-
-		time.Sleep(1 * time.Second)
-		r.FlushAll()
-	}
-
-	myControl.Stop()
-	theirControl.Stop()
-}

examples/config.yml

@@ -342,14 +342,6 @@ logging:
   # after receiving the response for lighthouse queries
   #trigger_buffer: 64
 
-  # max_rate limits the number of new inbound handshakes per second. Once the limit is reached,
-  # new handshakes are dropped until the next second. A value of 0 means unlimited (default).
-  # This is useful for preventing DoS attacks that attempt to exhaust CPU with handshake crypto.
-  # Running `openssl speed ecdhp256` on your hardware can be a good rule of thumb for choosing
-  # a max, as each handshake performs similar DH operations. Note that this benchmarks a single
-  # core, so you may wish to scale the value by the number of `routines` configured.
-  #max_rate: 0
-
 # Tunnel manager settings
 #tunnels:
   # drop_inactive controls whether inactive tunnels are maintained or dropped after the inactive_timeout period has

firewall.go

@@ -230,7 +230,7 @@ func NewFirewallFromConfig(l *logrus.Logger, cs *CertState, c *config.C) (*Firew
 	case "drop":
 		fw.OutSendReject = false
 	default:
-		l.WithField("action", outboundAction).Warn("invalid firewall.outbound_action, defaulting to `drop`")
+		l.WithField("action", inboundAction).Warn("invalid firewall.outbound_action, defaulting to `drop`")
 		fw.OutSendReject = false
 	}
 

handshake_manager.go

@@ -23,25 +23,22 @@ const (
 	DefaultHandshakeRetries       = 10
 	DefaultHandshakeTriggerBuffer = 64
 	DefaultUseRelays              = true
-	DefaultMaxHandshakeRate       = 0 // 0 means unlimited
 )
 
 var (
 	defaultHandshakeConfig = HandshakeConfig{
-		tryInterval:      DefaultHandshakeTryInterval,
+		tryInterval:   DefaultHandshakeTryInterval,
-		retries:          DefaultHandshakeRetries,
+		retries:       DefaultHandshakeRetries,
-		triggerBuffer:    DefaultHandshakeTriggerBuffer,
+		triggerBuffer: DefaultHandshakeTriggerBuffer,
-		useRelays:        DefaultUseRelays,
+		useRelays:     DefaultUseRelays,
-		maxHandshakeRate: DefaultMaxHandshakeRate,
 	}
 )
 
 type HandshakeConfig struct {
-	tryInterval      time.Duration
+	tryInterval   time.Duration
-	retries          int64
+	retries       int64
-	triggerBuffer    int
+	triggerBuffer int
-	useRelays        bool
+	useRelays     bool
-	maxHandshakeRate int
 
 	messageMetrics *MessageMetrics
 }
@@ -61,15 +58,9 @@ type HandshakeManager struct {
 	messageMetrics         *MessageMetrics
 	metricInitiated        metrics.Counter
 	metricTimedOut         metrics.Counter
-	metricRateLimited      metrics.Counter
 	f                      *Interface
 	l                      *logrus.Logger
 
-	// Rate limiting for new handshakes (token bucket)
-	rateBucket   int // tokens currently available
-	rateMax      int // max tokens (== max handshakes per second), 0 means unlimited
-	rateLastTick time.Time
-
 	// can be used to trigger outbound handshake for the given vpnIp
 	trigger chan netip.Addr
 }
@@ -125,41 +116,10 @@ func NewHandshakeManager(l *logrus.Logger, mainHostMap *HostMap, lightHouse *Lig
 		messageMetrics:         config.messageMetrics,
 		metricInitiated:        metrics.GetOrRegisterCounter("handshake_manager.initiated", nil),
 		metricTimedOut:         metrics.GetOrRegisterCounter("handshake_manager.timed_out", nil),
-		metricRateLimited:      metrics.GetOrRegisterCounter("handshake_manager.rate_limited", nil),
-		rateBucket:             config.maxHandshakeRate,
-		rateMax:                config.maxHandshakeRate,
-		rateLastTick:           time.Now(),
 		l:                      l,
 	}
 }
 
-// handshakeRateAllow checks the token bucket rate limiter and returns true if a
-// new handshake is allowed. Must be called with hm.Lock held.
-func (hm *HandshakeManager) handshakeRateAllow(now time.Time) bool {
-	if hm.rateMax == 0 {
-		return true
-	}
-
-	// Refill tokens based on elapsed time
-	elapsed := now.Sub(hm.rateLastTick)
-	if elapsed >= time.Second {
-		// Add tokens for full seconds elapsed
-		tokens := int(elapsed/time.Second) * hm.rateMax
-		hm.rateBucket += tokens
-		if hm.rateBucket > hm.rateMax {
-			hm.rateBucket = hm.rateMax
-		}
-		hm.rateLastTick = now
-	}
-
-	if hm.rateBucket > 0 {
-		hm.rateBucket--
-		return true
-	}
-
-	return false
-}
-
 func (hm *HandshakeManager) Run(ctx context.Context) {
 	clockSource := time.NewTicker(hm.config.tryInterval)
 	defer clockSource.Stop()
@@ -189,15 +149,6 @@ func (hm *HandshakeManager) HandleIncoming(via ViaSender, packet []byte, h *head
 	case header.HandshakeIXPSK0:
 		switch h.MessageCounter {
 		case 1:
-			// Check rate limit for new incoming handshakes
-			hm.Lock()
-			allowed := hm.handshakeRateAllow(time.Now())
-			hm.Unlock()
-			if !allowed {
-				hm.metricRateLimited.Inc(1)
-				hm.l.WithField("from", via).Debug("Handshake rate limit reached, dropping incoming handshake")
-				return
-			}
 			ixHandshakeStage1(hm.f, via, packet, h)
 
 		case 2:

handshake_manager_test.go

@@ -65,68 +65,6 @@ func Test_NewHandshakeManagerVpnIp(t *testing.T) {
 	assert.NotContains(t, blah.vpnIps, ip)
 }
 
-func Test_HandshakeManagerRateLimit(t *testing.T) {
-	l := test.NewLogger()
-	localrange := netip.MustParsePrefix("10.1.1.1/24")
-	preferredRanges := []netip.Prefix{localrange}
-	mainHM := newHostMap(l)
-	mainHM.preferredRanges.Store(&preferredRanges)
-
-	lh := newTestLighthouse()
-
-	config := defaultHandshakeConfig
-	config.maxHandshakeRate = 2
-
-	hm := NewHandshakeManager(l, mainHM, lh, &udp.NoopConn{}, config)
-	hm.f = &Interface{handshakeManager: hm, pki: &PKI{}, l: l}
-
-	now := time.Now()
-
-	// Should allow up to maxHandshakeRate handshakes
-	hm.Lock()
-	assert.True(t, hm.handshakeRateAllow(now), "first handshake should be allowed")
-	assert.True(t, hm.handshakeRateAllow(now), "second handshake should be allowed")
-	assert.False(t, hm.handshakeRateAllow(now), "third handshake should be rate limited")
-	hm.Unlock()
-
-	// After advancing time by 1 second, tokens should refill
-	hm.Lock()
-	assert.True(t, hm.handshakeRateAllow(now.Add(time.Second)), "handshake should be allowed after token refill")
-	hm.Unlock()
-}
-
-func Test_HandshakeManagerRateLimitUnlimited(t *testing.T) {
-	l := test.NewLogger()
-	localrange := netip.MustParsePrefix("10.1.1.1/24")
-	preferredRanges := []netip.Prefix{localrange}
-	mainHM := newHostMap(l)
-	mainHM.preferredRanges.Store(&preferredRanges)
-
-	lh := newTestLighthouse()
-
-	cs := &CertState{
-		initiatingVersion: cert.Version1,
-		privateKey:        []byte{},
-		v1Cert:            &dummyCert{version: cert.Version1},
-		v1HandshakeBytes:  []byte{},
-	}
-
-	// Default config has maxHandshakeRate=0 (unlimited)
-	hm := NewHandshakeManager(l, mainHM, lh, &udp.NoopConn{}, defaultHandshakeConfig)
-	hm.f = &Interface{handshakeManager: hm, pki: &PKI{}, l: l}
-	hm.f.pki.cs.Store(cs)
-
-	// Should allow many handshakes with no limit
-	// Limited to 10 due to test lighthouse query channel buffer
-	for i := 0; i < 10; i++ {
-		ip := netip.MustParseAddr("172.1.1.1").As16()
-		ip[15] = byte(i + 1)
-		addr := netip.AddrFrom16(ip)
-		h := hm.StartHandshake(addr, nil)
-		assert.NotNil(t, h, "handshake %d should be allowed with unlimited rate", i)
-	}
-}
-
 func testCountTimerWheelEntries(tw *LockingTimerWheel[netip.Addr]) (c int) {
 	for _, i := range tw.t.wheel {
 		n := i.Head

interface.go

@@ -263,8 +263,6 @@ func (f *Interface) run() {
 }
 
 func (f *Interface) listenOut(i int) {
-	runtime.LockOSThread()
-
 	var li udp.Conn
 	if i > 0 {
 		li = f.writers[i]

main.go

@@ -204,11 +204,10 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
 	useRelays := c.GetBool("relay.use_relays", DefaultUseRelays) && !c.GetBool("relay.am_relay", false)
 
 	handshakeConfig := HandshakeConfig{
-		tryInterval:      c.GetDuration("handshakes.try_interval", DefaultHandshakeTryInterval),
+		tryInterval:   c.GetDuration("handshakes.try_interval", DefaultHandshakeTryInterval),
-		retries:          int64(c.GetInt("handshakes.retries", DefaultHandshakeRetries)),
+		retries:       int64(c.GetInt("handshakes.retries", DefaultHandshakeRetries)),
-		triggerBuffer:    c.GetInt("handshakes.trigger_buffer", DefaultHandshakeTriggerBuffer),
+		triggerBuffer: c.GetInt("handshakes.trigger_buffer", DefaultHandshakeTriggerBuffer),
-		useRelays:        useRelays,
+		useRelays:     useRelays,
-		maxHandshakeRate: c.GetInt("handshakes.max_rate", DefaultMaxHandshakeRate),
 
 		messageMetrics: messageMetrics,
 	}

outside.go

@@ -190,13 +190,6 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
 		if !f.handleEncrypted(ci, via, h) {
 			return
 		}
-		_, err = f.decrypt(hostinfo, h.MessageCounter, out, packet, h, nb)
-		if err != nil {
-			hostinfo.logger(f.l).WithError(err).WithField("from", via).
-				WithField("packet", packet).
-				Error("Failed to decrypt CloseTunnel packet")
-			return
-		}
 
 		hostinfo.logger(f.l).WithField("from", via).
 			Info("Close tunnel received, tearing down.")

ssh.go

@@ -632,9 +632,6 @@ func sshCreateTunnel(ifce *Interface, fs any, a []string, w sshd.StringWriter) e
 	}
 
 	hostInfo = ifce.handshakeManager.StartHandshake(vpnAddr, nil)
-	if hostInfo == nil {
-		return w.WriteLine("Handshake rate limit reached")
-	}
 	if addr.IsValid() {
 		hostInfo.SetRemote(addr)
 	}

udp/udp_linux.go

@@ -4,6 +4,7 @@
 package udp
 
 import (
+	"context"
 	"encoding/binary"
 	"fmt"
 	"net"
@@ -18,58 +19,66 @@ import (
 )
 
 type StdConn struct {
-	sysFd int
+	udpConn *net.UDPConn
-	isV4  bool
+	rawConn syscall.RawConn
-	l     *logrus.Logger
+	sockFd  int
-	batch int
+	isV4    bool
+	l       *logrus.Logger
+	batch   int
 }
 
-func maybeIPV4(ip net.IP) (net.IP, bool) {
+func setReusePort(network, address string, c syscall.RawConn) error {
-	ip4 := ip.To4()
+	var opErr error
-	if ip4 != nil {
+	err := c.Control(func(fd uintptr) {
-		return ip4, true
+		opErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
+		//CloseOnExec already set by the runtime
+	})
+	if err != nil {
+		return err
 	}
-	return ip, false
+	return opErr
 }
 
 func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
-	af := unix.AF_INET6
+	listen := netip.AddrPortFrom(ip, uint16(port))
-	if ip.Is4() {
+	lc := net.ListenConfig{}
-		af = unix.AF_INET
+	if multi {
+		lc.Control = setReusePort
 	}
-	syscall.ForkLock.RLock()
+	network := "udp"
-	fd, err := unix.Socket(af, unix.SOCK_DGRAM, unix.IPPROTO_UDP)
+	if ip.Is4() { //todo I am not a fan of this, would prefer a default where we always bind dual-stack unless given a non-wildcard v4 addr
-	if err == nil {
+		network = "udp4"
-		unix.CloseOnExec(fd)
 	}
-	syscall.ForkLock.RUnlock()
+	//this context is only used during the bind operation, you can't cancel it to kill the socket
-
+	pc, err := lc.ListenPacket(context.Background(), network, listen.String())
 	if err != nil {
-		unix.Close(fd)
 		return nil, fmt.Errorf("unable to open socket: %s", err)
 	}
-
+	udpConn := pc.(*net.UDPConn)
-	if multi {
+	rawConn, err := udpConn.SyscallConn()
-		if err = unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEPORT, 1); err != nil {
+	if err != nil {
-			return nil, fmt.Errorf("unable to set SO_REUSEPORT: %s", err)
+		_ = udpConn.Close()
-		}
+		return nil, err
 	}
 
-	var sa unix.Sockaddr
+	//steal the socket's fd for sending
-	if ip.Is4() {
+	var sockFd uintptr
-		sa4 := &unix.SockaddrInet4{Port: port}
+	err = rawConn.Control(func(fd uintptr) {
-		sa4.Addr = ip.As4()
+		sockFd = fd
-		sa = sa4
+	})
-	} else {
+	if err != nil {
-		sa6 := &unix.SockaddrInet6{Port: port}
+		_ = udpConn.Close()
-		sa6.Addr = ip.As16()
+		return nil, err
-		sa = sa6
-	}
-	if err = unix.Bind(fd, sa); err != nil {
-		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{
+		udpConn: udpConn,
+		rawConn: rawConn,
+		sockFd:  int(sockFd),
+		isV4:    ip.Is4(),
+		l:       l,
+		batch:   batch,
+	}, err
 }
 
 func (u *StdConn) SupportsMultipleReaders() bool {
@@ -80,63 +89,126 @@ func (u *StdConn) Rebind() error {
 	return nil
 }
 
+func (u *StdConn) getSockOptInt(opt int) (int, error) {
+	if u.rawConn == nil {
+		return 0, fmt.Errorf("no UDP connection")
+	}
+	var out int
+	var opErr error
+	err := u.rawConn.Control(func(fd uintptr) {
+		out, opErr = unix.GetsockoptInt(int(fd), unix.SOL_SOCKET, opt)
+	})
+	if err != nil {
+		return 0, err
+	}
+	return out, opErr
+}
+
+func (u *StdConn) setSockOptInt(opt int, n int) error {
+	if u.rawConn == nil {
+		return fmt.Errorf("no UDP connection")
+	}
+	var opErr error
+	err := u.rawConn.Control(func(fd uintptr) {
+		opErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, opt, n)
+	})
+	if err != nil {
+		return err
+	}
+	return opErr
+}
+
 func (u *StdConn) SetRecvBuffer(n int) error {
-	return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_RCVBUFFORCE, n)
+	return u.setSockOptInt(unix.SO_RCVBUFFORCE, n)
 }
 
 func (u *StdConn) SetSendBuffer(n int) error {
-	return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_SNDBUFFORCE, n)
+	return u.setSockOptInt(unix.SO_SNDBUFFORCE, n)
 }
 
 func (u *StdConn) SetSoMark(mark int) error {
-	return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_MARK, mark)
+	return u.setSockOptInt(unix.SO_MARK, mark)
 }
 
 func (u *StdConn) GetRecvBuffer() (int, error) {
-	return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_RCVBUF)
+	return u.getSockOptInt(unix.SO_RCVBUF)
 }
 
 func (u *StdConn) GetSendBuffer() (int, error) {
-	return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_SNDBUF)
+	return u.getSockOptInt(unix.SO_SNDBUF)
 }
 
 func (u *StdConn) GetSoMark() (int, error) {
-	return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_MARK)
+	return u.getSockOptInt(unix.SO_MARK)
 }
 
 func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
-	sa, err := unix.Getsockname(u.sysFd)
+	addr := u.udpConn.LocalAddr()
-	if err != nil {
+	return netip.ParseAddrPort(addr.String())
-		return netip.AddrPort{}, err
+}
+
+func recvmmsg(fd uintptr, msgs []rawMessage) (int, bool, error) {
+	var errno syscall.Errno
+	n, _, errno := unix.Syscall6(
+		unix.SYS_RECVMMSG,
+		fd,
+		uintptr(unsafe.Pointer(&msgs[0])),
+		uintptr(len(msgs)),
+		unix.MSG_WAITFORONE,
+		0,
+		0,
+	)
+	if errno == syscall.EAGAIN || errno == syscall.EWOULDBLOCK {
+		// No data available, block for I/O and try again.
+		return int(n), false, nil
 	}
+	if errno != 0 {
+		return int(n), true, &net.OpError{Op: "recvmmsg", Err: errno}
+	}
+	return int(n), true, nil
+}
 
-	switch sa := sa.(type) {
+func (u *StdConn) listenOutSingle(r EncReader) {
-	case *unix.SockaddrInet4:
+	var err error
-		return netip.AddrPortFrom(netip.AddrFrom4(sa.Addr), uint16(sa.Port)), nil
+	var n int
+	var from netip.AddrPort
+	buffer := make([]byte, MTU)
 
-	case *unix.SockaddrInet6:
+	for {
-		return netip.AddrPortFrom(netip.AddrFrom16(sa.Addr), uint16(sa.Port)), nil
+		n, from, err = u.udpConn.ReadFromUDPAddrPort(buffer)
-
+		if err != nil {
-	default:
+			u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
-		return netip.AddrPort{}, fmt.Errorf("unsupported sock type: %T", sa)
+			return
+		}
+		from = netip.AddrPortFrom(from.Addr().Unmap(), from.Port())
+		r(from, buffer[:n])
 	}
 }
 
-func (u *StdConn) ListenOut(r EncReader) {
+func (u *StdConn) listenOutBatch(r EncReader) {
 	var ip netip.Addr
+	var n int
+	var operr error
 
 	msgs, buffers, names := u.PrepareRawMessages(u.batch)
-	read := u.ReadMulti
+
-	if u.batch == 1 {
+	//reader needs to capture variables from this function, since it's used as a lambda with rawConn.Read
-		read = u.ReadSingle
+	//defining it outside the loop so it gets re-used
+	reader := func(fd uintptr) (done bool) {
+		n, done, operr = recvmmsg(fd, msgs)
+		return done
 	}
 
 	for {
-		n, err := read(msgs)
+		err := u.rawConn.Read(reader)
 		if err != nil {
 			u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
 			return
 		}
+		if operr != nil {
+			u.l.WithError(operr).Debug("operr: udp socket is closed, exiting read loop")
+			return
+		}
 
 		for i := 0; i < n; i++ {
 			// Its ok to skip the ok check here, the slicing is the only error that can occur and it will panic
@@ -150,44 +222,14 @@ func (u *StdConn) ListenOut(r EncReader) {
 	}
 }
 
-func (u *StdConn) ReadSingle(msgs []rawMessage) (int, error) {
+func (u *StdConn) ListenOut(r EncReader) {
-	for {
+	if u.batch == 1 {
-		n, _, err := unix.Syscall6(
+		//save some ram by not calling PrepareRawMessages for fields we won't use
-			unix.SYS_RECVMSG,
+		//we could also make this path more common by calling recvmmsg with msgs[:1],
-			uintptr(u.sysFd),
+		//but that's still the recvmmsg syscall, which would be a change
-			uintptr(unsafe.Pointer(&(msgs[0].Hdr))),
+		u.listenOutSingle(r)
-			0,
+	} else {
-			0,
+		u.listenOutBatch(r)
-			0,
-			0,
-		)
-
-		if err != 0 {
-			return 0, &net.OpError{Op: "recvmsg", Err: err}
-		}
-
-		msgs[0].Len = uint32(n)
-		return 1, nil
-	}
-}
-
-func (u *StdConn) ReadMulti(msgs []rawMessage) (int, error) {
-	for {
-		n, _, err := unix.Syscall6(
-			unix.SYS_RECVMMSG,
-			uintptr(u.sysFd),
-			uintptr(unsafe.Pointer(&msgs[0])),
-			uintptr(len(msgs)),
-			unix.MSG_WAITFORONE,
-			0,
-			0,
-		)
-
-		if err != 0 {
-			return 0, &net.OpError{Op: "recvmmsg", Err: err}
-		}
-
-		return int(n), nil
 	}
 }
 
@@ -207,7 +249,7 @@ func (u *StdConn) writeTo6(b []byte, ip netip.AddrPort) error {
 	for {
 		_, _, err := unix.Syscall6(
 			unix.SYS_SENDTO,
-			uintptr(u.sysFd),
+			uintptr(u.sockFd),
 			uintptr(unsafe.Pointer(&b[0])),
 			uintptr(len(b)),
 			uintptr(0),
@@ -236,7 +278,7 @@ func (u *StdConn) writeTo4(b []byte, ip netip.AddrPort) error {
 	for {
 		_, _, err := unix.Syscall6(
 			unix.SYS_SENDTO,
-			uintptr(u.sysFd),
+			uintptr(u.sockFd),
 			uintptr(unsafe.Pointer(&b[0])),
 			uintptr(len(b)),
 			uintptr(0),
@@ -302,15 +344,28 @@ func (u *StdConn) ReloadConfig(c *config.C) {
 
 func (u *StdConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error {
 	var vallen uint32 = 4 * unix.SK_MEMINFO_VARS
-	_, _, err := unix.Syscall6(unix.SYS_GETSOCKOPT, uintptr(u.sysFd), uintptr(unix.SOL_SOCKET), uintptr(unix.SO_MEMINFO), uintptr(unsafe.Pointer(meminfo)), uintptr(unsafe.Pointer(&vallen)), 0)
+
-	if err != 0 {
+	if u.rawConn == nil {
+		return fmt.Errorf("no UDP connection")
+	}
+	var opErr error
+	err := u.rawConn.Control(func(fd uintptr) {
+		_, _, syserr := unix.Syscall6(unix.SYS_GETSOCKOPT, fd, uintptr(unix.SOL_SOCKET), uintptr(unix.SO_MEMINFO), uintptr(unsafe.Pointer(meminfo)), uintptr(unsafe.Pointer(&vallen)), 0)
+		if syserr != 0 {
+			opErr = syserr
+		}
+	})
+	if err != nil {
 		return err
 	}
-	return nil
+	return opErr
 }
 
 func (u *StdConn) Close() error {
-	return syscall.Close(u.sysFd)
+	if u.udpConn != nil {
+		return u.udpConn.Close()
+	}
+	return nil
 }
 
 func NewUDPStatsEmitter(udpConns []Conn) func() {