fixed tcp/udp I/O, deadlock, nil dereference; improved docker watcher, idlewatcher, loading page

src/route/constants.go

@@ -4,5 +4,5 @@ import (
 	"time"
 )
 
-const udpBufferSize = 1500
+const udpBufferSize = 8192
 const streamStopListenTimeout = 1 * time.Second

src/route/http_route.go

@@ -37,19 +37,23 @@ type (
 )
 
 func NewHTTPRoute(entry *P.ReverseProxyEntry) (*HTTPRoute, E.NestedError) {
-	var trans http.RoundTripper
+	var trans *http.Transport
 	var regIdleWatcher func() E.NestedError
 	var unregIdleWatcher func()
 
 	if entry.NoTLSVerify {
-		trans = transportNoTLS
+		trans = transportNoTLS.Clone()
 	} else {
-		trans = transport
+		trans = transport.Clone()
 	}
 
 	rp := P.NewReverseProxy(entry.URL, trans, entry)
 
 	if entry.UseIdleWatcher() {
+		// allow time for response header up to `WakeTimeout`
+		if entry.WakeTimeout > trans.ResponseHeaderTimeout {
+			trans.ResponseHeaderTimeout = entry.WakeTimeout
+		}
 		regIdleWatcher = func() E.NestedError {
 			watcher, err := idlewatcher.Register(entry)
 			if err.HasError() {
@@ -114,6 +118,7 @@ func (r *HTTPRoute) Stop() E.NestedError {
 
 	if r.unregIdleWatcher != nil {
 		r.unregIdleWatcher()
+		r.unregIdleWatcher = nil
 	}
 
 	r.mux = nil
@@ -151,13 +156,13 @@ func findMux(host string) (*http.ServeMux, E.NestedError) {
 }
 
 var (
+	defaultDialer = net.Dialer{
+		Timeout:   60 * time.Second,
+		KeepAlive: 60 * time.Second,
+	}
 	transport = &http.Transport{
-		Proxy: http.ProxyFromEnvironment,
-		DialContext: (&net.Dialer{
-			Timeout:   60 * time.Second,
-			KeepAlive: 60 * time.Second,
-		}).DialContext,
-		MaxIdleConns:        1000,
+		Proxy:               http.ProxyFromEnvironment,
+		DialContext:         defaultDialer.DialContext,
 		MaxIdleConnsPerHost: 1000,
 	}
 	transportNoTLS = func() *http.Transport {

src/route/stream_route.go

@@ -2,6 +2,7 @@ package route
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -129,7 +130,7 @@ func (r *StreamRoute) grHandleConnections() {
 		case conn := <-r.connCh:
 			go func() {
 				err := r.Handle(conn)
-				if err != nil {
+				if err != nil && !errors.Is(err, context.Canceled) {
 					r.l.Error(err)
 				}
 			}()

src/route/tcp_route.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"net"
 	"sync"
-	"syscall"
 	"time"
 
 	U "github.com/yusing/go-proxy/utils"
@@ -13,14 +12,16 @@ import (
 
 const tcpDialTimeout = 5 * time.Second
 
-type Pipes []*U.BidirectionalPipe
+type (
+	Pipes []U.BidirectionalPipe
 
-type TCPRoute struct {
-	*StreamRoute
-	listener net.Listener
-	pipe     Pipes
-	mu       sync.Mutex
-}
+	TCPRoute struct {
+		*StreamRoute
+		listener net.Listener
+		pipe     Pipes
+		mu       sync.Mutex
+	}
+)
 
 func NewTCPRoute(base *StreamRoute) StreamImpl {
 	return &TCPRoute{
@@ -59,10 +60,11 @@ func (route *TCPRoute) Handle(c any) error {
 	}
 
 	route.mu.Lock()
-	defer route.mu.Unlock()
 
 	pipe := U.NewBidirectionalPipe(route.ctx, clientConn, serverConn)
 	route.pipe = append(route.pipe, pipe)
+
+	route.mu.Unlock()
 	return pipe.Start()
 }
 
@@ -72,16 +74,4 @@ func (route *TCPRoute) CloseListeners() {
 	}
 	route.listener.Close()
 	route.listener = nil
-	for _, pipe := range route.pipe {
-		if err := pipe.Stop(); err != nil {
-			switch err {
-			// target closing connection
-			// TODO: handle this by fixing utils/io.go
-			case net.ErrClosed, syscall.EPIPE:
-				return
-			default:
-				route.l.Error(err)
-			}
-		}
-	}
 }

src/route/udp_route.go

@@ -4,33 +4,34 @@ import (
 	"fmt"
 	"io"
 	"net"
-	"sync"
 
-	"github.com/yusing/go-proxy/utils"
+	U "github.com/yusing/go-proxy/utils"
+	F "github.com/yusing/go-proxy/utils/functional"
 )
 
-type UDPRoute struct {
-	*StreamRoute
+type (
+	UDPRoute struct {
+		*StreamRoute
 
-	connMap      UDPConnMap
-	connMapMutex sync.Mutex
+		connMap UDPConnMap
 
-	listeningConn *net.UDPConn
-	targetAddr    *net.UDPAddr
-}
+		listeningConn *net.UDPConn
+		targetAddr    *net.UDPAddr
+	}
+	UDPConn struct {
+		src *net.UDPConn
+		dst *net.UDPConn
+		U.BidirectionalPipe
+	}
+	UDPConnMap = F.Map[string, *UDPConn]
+)
 
-type UDPConn struct {
-	src *net.UDPConn
-	dst *net.UDPConn
-	*utils.BidirectionalPipe
-}
-
-type UDPConnMap map[string]*UDPConn
+var NewUDPConnMap = F.NewMapOf[string, *UDPConn]
 
 func NewUDPRoute(base *StreamRoute) StreamImpl {
 	return &UDPRoute{
 		StreamRoute: base,
-		connMap:     make(UDPConnMap),
+		connMap:     NewUDPConnMap(),
 	}
 }
 
@@ -69,28 +70,24 @@ func (route *UDPRoute) Accept() (any, error) {
 	}
 
 	key := srcAddr.String()
-	conn, ok := route.connMap[key]
+	conn, ok := route.connMap.Load(key)
 
 	if !ok {
-		route.connMapMutex.Lock()
-		if conn, ok = route.connMap[key]; !ok {
-			srcConn, err := net.DialUDP("udp", nil, srcAddr)
-			if err != nil {
-				return nil, err
-			}
-			dstConn, err := net.DialUDP("udp", nil, route.targetAddr)
-			if err != nil {
-				srcConn.Close()
-				return nil, err
-			}
-			conn = &UDPConn{
-				srcConn,
-				dstConn,
-				utils.NewBidirectionalPipe(route.ctx, sourceRWCloser{in, dstConn}, sourceRWCloser{in, srcConn}),
-			}
-			route.connMap[key] = conn
+		srcConn, err := net.DialUDP("udp", nil, srcAddr)
+		if err != nil {
+			return nil, err
 		}
-		route.connMapMutex.Unlock()
+		dstConn, err := net.DialUDP("udp", nil, route.targetAddr)
+		if err != nil {
+			srcConn.Close()
+			return nil, err
+		}
+		conn = &UDPConn{
+			srcConn,
+			dstConn,
+			U.NewBidirectionalPipe(route.ctx, sourceRWCloser{in, dstConn}, sourceRWCloser{in, srcConn}),
+		}
+		route.connMap.Store(key, conn)
 	}
 
 	_, err = conn.dst.Write(buffer[:nRead])
@@ -106,15 +103,15 @@ func (route *UDPRoute) CloseListeners() {
 		route.listeningConn.Close()
 		route.listeningConn = nil
 	}
-	for _, conn := range route.connMap {
+	route.connMap.RangeAll(func(_ string, conn *UDPConn) {
 		if err := conn.src.Close(); err != nil {
 			route.l.Errorf("error closing src conn: %s", err)
 		}
 		if err := conn.dst.Close(); err != nil {
 			route.l.Error("error closing dst conn: %s", err)
 		}
-	}
-	route.connMap = make(UDPConnMap)
+	})
+	route.connMap.Clear()
 }
 
 type sourceRWCloser struct {