Merge pull request #1436 from apernet/fix-race

fix: race in udpSessionManager cleanup & update code formatting

Author: tobyxdd

app/internal/redirect/getsockopt_linux.go

@@ -13,5 +13,5 @@ func getsockopt(s, level, name uintptr, val unsafe.Pointer, vallen *uint32) (err
 	if e != 0 {
 		err = e
 	}
-	return
+	return err
 }

app/internal/redirect/getsockopt_linux_386.go

@@ -19,5 +19,5 @@ func getsockopt(s, level, name uintptr, val unsafe.Pointer, vallen *uint32) (err
 	if e != 0 {
 		err = e
 	}
-	return
+	return err
 }

app/internal/sockopts/sockopts.go

@@ -42,15 +42,15 @@ func (e *UnsupportedError) Error() string {
 func (o *SocketOptions) ListenUDP() (uconn net.PacketConn, err error) {
 	uconn, err = net.ListenUDP("udp", nil)
 	if err != nil {
-		return
+		return uconn, err
 	}
 	err = o.applyToUDPConn(uconn.(*net.UDPConn))
 	if err != nil {
 		uconn.Close()
 		uconn = nil
-		return
+		return uconn, err
 	}
-	return
+	return uconn, err
 }
 
 func (o *SocketOptions) applyToUDPConn(c *net.UDPConn) error {

app/internal/sockopts/sockopts_linux.go

@@ -24,19 +24,19 @@ func init() {
 func controlUDPConn(c *net.UDPConn, cb func(fd int) error) (err error) {
 	rconn, err := c.SyscallConn()
 	if err != nil {
-		return
+		return err
 	}
 	cerr := rconn.Control(func(fd uintptr) {
 		err = cb(int(fd))
 	})
 	if err != nil {
-		return
+		return err
 	}
 	if cerr != nil {
 		err = fmt.Errorf("failed to control fd: %w", cerr)
-		return
+		return err
 	}
-	return
+	return err
 }
 
 func bindInterfaceImpl(c *net.UDPConn, device string) error {

app/internal/sockopts/sockopts_linux_test.go

@@ -42,12 +42,12 @@ func Test_fdControlUnixSocketImpl(t *testing.T) {
 	err = controlUDPConn(conn.(*net.UDPConn), func(fd int) (err error) {
 		rcvbuf, err := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_RCVBUF)
 		if err != nil {
-			return
+			return err
 		}
 		// The test server called setsockopt(fd, SOL_SOCKET, SO_RCVBUF, 2500),
 		// and kernel will double this value for getsockopt().
 		assert.Equal(t, 5000, rcvbuf)
-		return
+		return err
 	})
 	assert.NoError(t, err)
 }

app/internal/url/url.go

@@ -1173,7 +1173,7 @@ func splitHostPort(hostPort string) (host, port string) {
 		host = host[1 : len(host)-1]
 	}
 
-	return
+	return host, port
 }
 
 // Marshaling interface implementations.
@@ -1263,8 +1263,8 @@ func stringContainsCTLByte(s string) bool {
 func JoinPath(base string, elem ...string) (result string, err error) {
 	url, err := Parse(base)
 	if err != nil {
-		return
+		return result, err
 	}
 	result = url.JoinPath(elem...).String()
-	return
+	return result, err
 }

app/internal/utils/certloader.go

@@ -68,42 +68,42 @@ func (l *LocalCertificateLoader) checkModTime() (certModTime, keyModTime time.Ti
 	fi, err := os.Stat(l.CertFile)
 	if err != nil {
 		err = fmt.Errorf("failed to stat certificate file: %w", err)
-		return
+		return certModTime, keyModTime, err
 	}
 	certModTime = fi.ModTime()
 
 	fi, err = os.Stat(l.KeyFile)
 	if err != nil {
 		err = fmt.Errorf("failed to stat key file: %w", err)
-		return
+		return certModTime, keyModTime, err
 	}
 	keyModTime = fi.ModTime()
-	return
+	return certModTime, keyModTime, err
 }
 
 func (l *LocalCertificateLoader) makeCache() (cache *localCertificateCache, err error) {
 	c := &localCertificateCache{}
 
 	c.certModTime, c.keyModTime, err = l.checkModTime()
 	if err != nil {
-		return
+		return cache, err
 	}
 
 	cert, err := tls.LoadX509KeyPair(l.CertFile, l.KeyFile)
 	if err != nil {
-		return
+		return cache, err
 	}
 	c.certificate = &cert
 	if c.certificate.Leaf == nil {
 		// certificate.Leaf was left nil by tls.LoadX509KeyPair before Go 1.23
 		c.certificate.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
 		if err != nil {
-			return
+			return cache, err
 		}
 	}
 
 	cache = c
-	return
+	return cache, err
 }
 
 func (l *LocalCertificateLoader) getCertificateWithCache() (*tls.Certificate, error) {

core/server/udp.go

@@ -60,7 +60,7 @@ func newUDPSessionEntry(
 		ExitFunc: exitFunc,
 	}
 
-	return
+	return e
 }
 
 // CloseWithErr closes the session and calls ExitFunc with the given error.
@@ -259,10 +259,9 @@ func (m *udpSessionManager) idleCleanupLoop(stopCh <-chan struct{}) {
 }
 
 func (m *udpSessionManager) cleanup(idleOnly bool) {
-	timeoutEntry := make([]*udpSessionEntry, 0, len(m.m))
-
 	// We use RLock here as we are only scanning the map, not deleting from it.
 	m.mutex.RLock()
+	timeoutEntry := make([]*udpSessionEntry, 0, len(m.m))
 	now := time.Now()
 	for _, entry := range m.m {
 		if !idleOnly || now.Sub(entry.Last.Get()) > m.idleTimeout {
@@ -289,14 +288,14 @@ func (m *udpSessionManager) feed(msg *protocol.UDPMessage) {
 			// Call the hook
 			err = m.io.Hook(firstMsgData, &addr)
 			if err != nil {
-				return
+				return conn, actualAddr, err
 			}
 			actualAddr = addr
 			// Log the event
 			m.eventLogger.New(msg.SessionID, addr)
 			// Dial target
 			conn, err = m.io.UDP(addr)
-			return
+			return conn, actualAddr, err
 		}
 		exitFunc := func(err error) {
 			// Log the event

extras/obfs/conn.go

@@ -64,12 +64,12 @@ func (c *obfsPacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
 		n, addr, err = c.Conn.ReadFrom(c.readBuf)
 		if n <= 0 {
 			c.readMutex.Unlock()
-			return
+			return n, addr, err
 		}
 		n = c.Obfs.Deobfuscate(c.readBuf[:n], p)
 		c.readMutex.Unlock()
 		if n > 0 || err != nil {
-			return
+			return n, addr, err
 		}
 		// Invalid packet, try again
 	}
@@ -83,7 +83,7 @@ func (c *obfsPacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
 	if err == nil {
 		n = len(p)
 	}
-	return
+	return n, err
 }
 
 func (c *obfsPacketConn) Close() error {

extras/outbounds/ob_socks5.go

@@ -258,7 +258,7 @@ func addrExToSOCKS5Addr(addr *AddrEx) (atyp byte, dstAddr, dstPort []byte) {
 	// Port
 	dstPort = make([]byte, 2)
 	binary.BigEndian.PutUint16(dstPort, addr.Port)
-	return
+	return atyp, dstAddr, dstPort
 }
 
 func socks5AddrToAddrEx(atyp byte, dstAddr, dstPort []byte) *AddrEx {