refactor: organize code (#90)

* fix: improved sync.Pool handling

* refactor: ping if-flow and remove timeout

* refactor: enhance favicon fetching with context support and improve cache management

- Added context support to favicon fetching functions to handle timeouts and cancellations.
- Improved cache entry structure to include content type and utilize atomic values for last access time.
- Implemented maximum cache size and entry limits to optimize memory usage.
- Updated error handling for HTTP requests and refined the logic for managing redirects.

* fix: log formatting

* feat(pool): add checkExists method to debug build to detect unexpected behavior

* chore: cont. 0866feb

* refactor: unify route handling by consolidating route query methods with Pool

- Replaced direct calls to routequery with a new routes package for better organization and maintainability.
- Updated various components to utilize the new routes methods for fetching health information, homepage configurations, and route aliases.
- Enhanced the overall structure of the routing logic to improve clarity and reduce redundancy.

* chore: uncomment icon list cache code

* refactor: update task management code

- Rename needFinish to waitFinish
- Fixed some tasks not being waited they should be
- Adjusted mutex usage in the directory watcher to utilize read-write locks for improved concurrency management.

* refactor: enhance idlewatcher logging and exit handling

* fix(server): ensure HTTP handler is set only if initialized

* refactor(accesslog): replace JSON log entry struct with zerolog for improved logging efficiency, updated test

* refactor: remove test run code

---------

Co-authored-by: yusing <yusing@6uo.me>

agent/pkg/agent/agents.go

@@ -10,7 +10,7 @@ var Agents = agents{pool.New[*AgentConfig]("agents")}
 
 func (agents agents) Get(agentAddrOrDockerHost string) (*AgentConfig, bool) {
 	if !IsDockerHostAgent(agentAddrOrDockerHost) {
-		return agents.Base().Load(agentAddrOrDockerHost)
+		return agents.Get(agentAddrOrDockerHost)
 	}
-	return agents.Base().Load(GetAgentAddrFromDockerHost(agentAddrOrDockerHost))
+	return agents.Get(GetAgentAddrFromDockerHost(agentAddrOrDockerHost))
 }

cmd/main.go

@@ -18,7 +18,7 @@ import (
 	"github.com/yusing/go-proxy/internal/metrics/systeminfo"
 	"github.com/yusing/go-proxy/internal/metrics/uptime"
 	"github.com/yusing/go-proxy/internal/net/gphttp/middleware"
-	"github.com/yusing/go-proxy/internal/route/routes/routequery"
+	"github.com/yusing/go-proxy/internal/route/routes"
 	"github.com/yusing/go-proxy/internal/task"
 	"github.com/yusing/go-proxy/migrations"
 	"github.com/yusing/go-proxy/pkg"
@@ -124,7 +124,7 @@ func main() {
 	switch args.Command {
 	case common.CommandListRoutes:
 		cfg.StartProxyProviders()
-		printJSON(routequery.RoutesByAlias())
+		printJSON(routes.ByAlias())
 		return
 	case common.CommandListConfigs:
 		printJSON(cfg.Value())

internal/api/v1/favicon/favicon.go

@@ -36,7 +36,7 @@ func GetFavIcon(w http.ResponseWriter, req *http.Request) {
 			gphttp.ClientError(w, err, http.StatusBadRequest)
 			return
 		}
-		fetchResult := homepage.FetchFavIconFromURL(&iconURL)
+		fetchResult := homepage.FetchFavIconFromURL(req.Context(), &iconURL)
 		if !fetchResult.OK() {
 			http.Error(w, fetchResult.ErrMsg, fetchResult.StatusCode)
 			return
@@ -47,7 +47,7 @@ func GetFavIcon(w http.ResponseWriter, req *http.Request) {
 	}
 
 	// try with route.Icon
-	r, ok := routes.GetHTTPRoute(alias)
+	r, ok := routes.HTTP.Get(alias)
 	if !ok {
 		gphttp.ClientError(w, errors.New("no such route"), http.StatusNotFound)
 		return
@@ -59,7 +59,7 @@ func GetFavIcon(w http.ResponseWriter, req *http.Request) {
 		if hp.Icon.IconSource == homepage.IconSourceRelative {
 			result = homepage.FindIcon(req.Context(), r, hp.Icon.Value)
 		} else {
-			result = homepage.FetchFavIconFromURL(hp.Icon)
+			result = homepage.FetchFavIconFromURL(req.Context(), hp.Icon)
 		}
 	} else {
 		// try extract from "link[rel=icon]"

internal/api/v1/health.go

@@ -5,9 +5,9 @@ import (
 	"time"
 
 	"github.com/yusing/go-proxy/internal/net/gphttp/gpwebsocket"
-	"github.com/yusing/go-proxy/internal/route/routes/routequery"
+	"github.com/yusing/go-proxy/internal/route/routes"
 )
 
 func Health(w http.ResponseWriter, r *http.Request) {
-	gpwebsocket.DynamicJSONHandler(w, r, routequery.HealthMap, 1*time.Second)
+	gpwebsocket.DynamicJSONHandler(w, r, routes.HealthMap, 1*time.Second)
 }

internal/api/v1/list.go

@@ -11,7 +11,7 @@ import (
 	"github.com/yusing/go-proxy/internal/homepage"
 	"github.com/yusing/go-proxy/internal/net/gphttp"
 	"github.com/yusing/go-proxy/internal/net/gphttp/middleware"
-	"github.com/yusing/go-proxy/internal/route/routes/routequery"
+	"github.com/yusing/go-proxy/internal/route/routes"
 	route "github.com/yusing/go-proxy/internal/route/types"
 	"github.com/yusing/go-proxy/internal/utils"
 )
@@ -45,7 +45,7 @@ func List(cfg config.ConfigInstance, w http.ResponseWriter, r *http.Request) {
 			gphttp.RespondJSON(w, r, route)
 		}
 	case ListRoutes:
-		gphttp.RespondJSON(w, r, routequery.RoutesByAlias(route.RouteType(r.FormValue("type"))))
+		gphttp.RespondJSON(w, r, routes.ByAlias(route.RouteType(r.FormValue("type"))))
 	case ListFiles:
 		listFiles(w, r)
 	case ListMiddlewares:
@@ -55,11 +55,11 @@ func List(cfg config.ConfigInstance, w http.ResponseWriter, r *http.Request) {
 	case ListMatchDomains:
 		gphttp.RespondJSON(w, r, cfg.Value().MatchDomains)
 	case ListHomepageConfig:
-		gphttp.RespondJSON(w, r, routequery.HomepageConfig(r.FormValue("category"), r.FormValue("provider")))
+		gphttp.RespondJSON(w, r, routes.HomepageConfig(r.FormValue("category"), r.FormValue("provider")))
 	case ListRouteProviders:
 		gphttp.RespondJSON(w, r, cfg.RouteProviderList())
 	case ListHomepageCategories:
-		gphttp.RespondJSON(w, r, routequery.HomepageCategories())
+		gphttp.RespondJSON(w, r, routes.HomepageCategories())
 	case ListIcons:
 		limit, err := strconv.Atoi(r.FormValue("limit"))
 		if err != nil {
@@ -83,9 +83,9 @@ func List(cfg config.ConfigInstance, w http.ResponseWriter, r *http.Request) {
 // otherwise, return a single Route with alias which or nil if not found.
 func listRoute(which string) any {
 	if which == "" || which == "all" {
-		return routequery.RoutesByAlias()
+		return routes.ByAlias()
 	}
-	routes := routequery.RoutesByAlias()
+	routes := routes.ByAlias()
 	route, ok := routes[which]
 	if !ok {
 		return nil

internal/docker/client.go

@@ -46,7 +46,7 @@ const (
 )
 
 func initClientCleaner() {
-	cleaner := task.RootTask("docker_clients_cleaner")
+	cleaner := task.RootTask("docker_clients_cleaner", false)
 	go func() {
 		ticker := time.NewTicker(cleanInterval)
 		defer ticker.Stop()

internal/entrypoint/entrypoint.go

@@ -12,7 +12,6 @@ import (
 	"github.com/yusing/go-proxy/internal/net/gphttp/middleware"
 	"github.com/yusing/go-proxy/internal/net/gphttp/middleware/errorpage"
 	"github.com/yusing/go-proxy/internal/route/routes"
-	route "github.com/yusing/go-proxy/internal/route/types"
 	"github.com/yusing/go-proxy/internal/task"
 	"github.com/yusing/go-proxy/internal/utils/strutils"
 )
@@ -20,7 +19,7 @@ import (
 type Entrypoint struct {
 	middleware    *middleware.Middleware
 	accessLogger  *accesslog.AccessLogger
-	findRouteFunc func(host string) (route.HTTPRoute, error)
+	findRouteFunc func(host string) (routes.HTTPRoute, error)
 }
 
 var ErrNoSuchRoute = errors.New("no such route")
@@ -108,7 +107,7 @@ func (ep *Entrypoint) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func findRouteAnyDomain(host string) (route.HTTPRoute, error) {
+func findRouteAnyDomain(host string) (routes.HTTPRoute, error) {
 	hostSplit := strutils.SplitRune(host, '.')
 	target := hostSplit[0]
 
@@ -118,19 +117,19 @@ func findRouteAnyDomain(host string) (route.HTTPRoute, error) {
 	return nil, fmt.Errorf("%w: %s", ErrNoSuchRoute, target)
 }
 
-func findRouteByDomains(domains []string) func(host string) (route.HTTPRoute, error) {
+func findRouteByDomains(domains []string) func(host string) (routes.HTTPRoute, error) {
-	return func(host string) (route.HTTPRoute, error) {
+	return func(host string) (routes.HTTPRoute, error) {
 		for _, domain := range domains {
 			if strings.HasSuffix(host, domain) {
 				target := strings.TrimSuffix(host, domain)
-				if r, ok := routes.GetHTTPRoute(target); ok {
+				if r, ok := routes.HTTP.Get(target); ok {
 					return r, nil
 				}
 			}
 		}
 
 		// fallback to exact match
-		if r, ok := routes.GetHTTPRoute(host); ok {
+		if r, ok := routes.HTTP.Get(host); ok {
 			return r, nil
 		}
 		return nil, fmt.Errorf("%w: %s", ErrNoSuchRoute, host)

internal/entrypoint/entrypoint_test.go

@@ -8,21 +8,29 @@ import (
 	. "github.com/yusing/go-proxy/internal/utils/testing"
 )
 
-var (
+var ep = NewEntrypoint()
-	r  route.ReveseProxyRoute
+
-	ep = NewEntrypoint()
+func addRoute(alias string) *route.ReveseProxyRoute {
-)
+	r := &route.ReveseProxyRoute{
+		Route: &route.Route{
+			Alias: alias,
+		},
+	}
+	routes.HTTP.Add(r)
+	return r
+}
 
 func run(t *testing.T, match []string, noMatch []string) {
 	t.Helper()
-	t.Cleanup(routes.TestClear)
+	t.Cleanup(routes.Clear)
 	t.Cleanup(func() { ep.SetFindRouteDomains(nil) })
 
 	for _, test := range match {
 		t.Run(test, func(t *testing.T) {
+			r := addRoute(test)
 			found, err := ep.findRouteFunc(test)
 			ExpectNoError(t, err)
-			ExpectTrue(t, found == &r)
+			ExpectTrue(t, found == r)
 		})
 	}
 
@@ -35,7 +43,7 @@ func run(t *testing.T, match []string, noMatch []string) {
 }
 
 func TestFindRouteAnyDomain(t *testing.T) {
-	routes.SetHTTPRoute("app1", &r)
+	addRoute("app1")
 
 	tests := []string{
 		"app1.com",
@@ -66,7 +74,7 @@ func TestFindRouteExactHostMatch(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		routes.SetHTTPRoute(test, &r)
+		addRoute(test)
 	}
 
 	run(t, tests, testsNoMatch)
@@ -78,7 +86,7 @@ func TestFindRouteByDomains(t *testing.T) {
 		".sub.domain.com",
 	})
 
-	routes.SetHTTPRoute("app1", &r)
+	addRoute("app1")
 
 	tests := []string{
 		"app1.domain.com",
@@ -103,7 +111,7 @@ func TestFindRouteByDomainsExactMatch(t *testing.T) {
 		".sub.domain.com",
 	})
 
-	routes.SetHTTPRoute("app1.foo.bar", &r)
+	addRoute("app1")
 
 	tests := []string{
 		"app1.foo.bar", // exact match

internal/gperr/log.go

@@ -6,14 +6,14 @@ import (
 	"github.com/yusing/go-proxy/internal/logging"
 )
 
-func log(_ string, err error, level zerolog.Level, logger ...*zerolog.Logger) {
+func log(msg string, err error, level zerolog.Level, logger ...*zerolog.Logger) {
 	var l *zerolog.Logger
 	if len(logger) > 0 {
 		l = logger[0]
 	} else {
 		l = logging.GetLogger()
 	}
-	l.WithLevel(level).Msg(err.Error())
+	l.WithLevel(level).Msg(New(highlight(msg)).With(err).Error())
 }
 
 func LogFatal(msg string, err error, logger ...*zerolog.Logger) {

internal/gperr/multiline_test.go

@@ -7,15 +7,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestMultiline(t *testing.T) {
-	multiline := Multiline()
-	multiline.Addf("line 1 %s", "test")
-	multiline.Adds("line 2")
-	multiline.AddLines([]any{1, "2", 3.0, net.IPv4(127, 0, 0, 1)})
-	t.Error(New("result").With(multiline))
-	t.Error(multiline.Subject("subject").Withf("inner"))
-}
-
 func TestWrapMultiline(t *testing.T) {
 	multiline := Multiline()
 	var wrapper error = wrap(multiline)

internal/homepage/favicon.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"slices"
 	"strings"
 	"time"
 
@@ -24,8 +25,10 @@ type FetchResult struct {
 	contentType string
 }
 
+const faviconFetchTimeout = 3 * time.Second
+
 func (res *FetchResult) OK() bool {
-	return res.Icon != nil
+	return len(res.Icon) > 0
 }
 
 func (res *FetchResult) ContentType() string {
@@ -40,39 +43,55 @@ func (res *FetchResult) ContentType() string {
 
 const maxRedirectDepth = 5
 
-func FetchFavIconFromURL(iconURL *IconURL) *FetchResult {
+func FetchFavIconFromURL(ctx context.Context, iconURL *IconURL) *FetchResult {
 	switch iconURL.IconSource {
 	case IconSourceAbsolute:
-		return fetchIconAbsolute(iconURL.URL())
+		return fetchIconAbsolute(ctx, iconURL.URL())
 	case IconSourceRelative:
 		return &FetchResult{StatusCode: http.StatusBadRequest, ErrMsg: "unexpected relative icon"}
 	case IconSourceWalkXCode, IconSourceSelfhSt:
-		return fetchKnownIcon(iconURL)
+		return fetchKnownIcon(ctx, iconURL)
 	}
 	return &FetchResult{StatusCode: http.StatusBadRequest, ErrMsg: "invalid icon source"}
 }
 
-func fetchIconAbsolute(url string) *FetchResult {
+func fetchIconAbsolute(ctx context.Context, url string) *FetchResult {
 	if result := loadIconCache(url); result != nil {
 		return result
 	}
 
-	resp, err := gphttp.Get(url)
+	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
-	if err != nil || resp.StatusCode != http.StatusOK {
+	if err != nil {
-		if err == nil {
+		if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
-			err = errors.New(resp.Status)
+			return &FetchResult{StatusCode: http.StatusBadGateway, ErrMsg: "request timeout"}
 		}
+		return &FetchResult{StatusCode: http.StatusInternalServerError, ErrMsg: err.Error()}
+	}
+
+	resp, err := gphttp.Do(req)
+	if err == nil {
+		defer resp.Body.Close()
+	}
+	if err != nil || resp.StatusCode != http.StatusOK {
 		return &FetchResult{StatusCode: http.StatusBadGateway, ErrMsg: "connection error"}
 	}
 
-	defer resp.Body.Close()
 	icon, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return &FetchResult{StatusCode: http.StatusInternalServerError, ErrMsg: "internal error"}
 	}
 
-	storeIconCache(url, icon)
+	if len(icon) == 0 {
-	return &FetchResult{Icon: icon}
+		return &FetchResult{StatusCode: http.StatusNotFound, ErrMsg: "empty icon"}
+	}
+
+	res := &FetchResult{Icon: icon}
+	if contentType := resp.Header.Get("Content-Type"); contentType != "" {
+		res.contentType = contentType
+	}
+	// else leave it empty
+	storeIconCache(url, res)
+	return res
 }
 
 var nameSanitizer = strings.NewReplacer(
@@ -86,44 +105,53 @@ func sanitizeName(name string) string {
 	return strings.ToLower(nameSanitizer.Replace(name))
 }
 
-func fetchKnownIcon(url *IconURL) *FetchResult {
+func fetchKnownIcon(ctx context.Context, url *IconURL) *FetchResult {
 	// if icon isn't in the list, no need to fetch
 	if !url.HasIcon() {
 		return &FetchResult{StatusCode: http.StatusNotFound, ErrMsg: "no such icon"}
 	}
 
-	return fetchIconAbsolute(url.URL())
+	return fetchIconAbsolute(ctx, url.URL())
 }
 
-func fetchIcon(filetype, filename string) *FetchResult {
+func fetchIcon(ctx context.Context, filetype, filename string) *FetchResult {
-	result := fetchKnownIcon(NewSelfhStIconURL(filename, filetype))
+	result := fetchKnownIcon(ctx, NewSelfhStIconURL(filename, filetype))
-	if result.Icon == nil {
+	if result.OK() {
 		return result
 	}
-	return fetchKnownIcon(NewWalkXCodeIconURL(filename, filetype))
+	return fetchKnownIcon(ctx, NewWalkXCodeIconURL(filename, filetype))
 }
 
 func FindIcon(ctx context.Context, r route, uri string) *FetchResult {
-	key := routeKey(r)
+	if result := loadIconCache(r.Key()); result != nil {
-	if result := loadIconCache(key); result != nil {
 		return result
 	}
 
-	result := fetchIcon("png", sanitizeName(r.Reference()))
+	result := fetchIcon(ctx, "png", sanitizeName(r.Reference()))
 	if !result.OK() {
 		if r, ok := r.(httpRoute); ok {
 			// fallback to parse html
-			result = findIconSlow(ctx, r, uri, 0)
+			result = findIconSlow(ctx, r, uri, nil)
 		}
 	}
 	if result.OK() {
-		storeIconCache(key, result.Icon)
+		storeIconCache(r.Key(), result)
 	}
 	return result
 }
 
-func findIconSlow(ctx context.Context, r httpRoute, uri string, depth int) *FetchResult {
+func findIconSlow(ctx context.Context, r httpRoute, uri string, stack []string) *FetchResult {
-	ctx, cancel := context.WithTimeoutCause(ctx, 3*time.Second, errors.New("favicon request timeout"))
+	select {
+	case <-ctx.Done():
+		return &FetchResult{StatusCode: http.StatusBadGateway, ErrMsg: "request timeout"}
+	default:
+	}
+
+	if len(stack) > maxRedirectDepth {
+		return &FetchResult{StatusCode: http.StatusBadGateway, ErrMsg: "too many redirects"}
+	}
+
+	ctx, cancel := context.WithTimeoutCause(ctx, faviconFetchTimeout, errors.New("favicon request timeout"))
 	defer cancel()
 
 	newReq, err := http.NewRequestWithContext(ctx, "GET", r.TargetURL().String(), nil)
@@ -149,14 +177,13 @@ func findIconSlow(ctx context.Context, r httpRoute, uri string, depth int) *Fetc
 			return &FetchResult{StatusCode: http.StatusBadGateway, ErrMsg: "connection error"}
 		default:
 			if loc := c.Header().Get("Location"); loc != "" {
-				if depth > maxRedirectDepth {
-					return &FetchResult{StatusCode: http.StatusBadGateway, ErrMsg: "too many redirects"}
-				}
 				loc = strutils.SanitizeURI(loc)
-				if loc == "/" || loc == newReq.URL.Path {
+				if loc == "/" || loc == newReq.URL.Path || slices.Contains(stack, loc) {
 					return &FetchResult{StatusCode: http.StatusBadGateway, ErrMsg: "circular redirect"}
 				}
-				return findIconSlow(ctx, r, loc, depth+1)
+				// append current path to stack
+				// handles redirect to the same path with different query
+				return findIconSlow(ctx, r, loc, append(stack, newReq.URL.Path))
 			}
 		}
 		return &FetchResult{StatusCode: c.status, ErrMsg: "upstream error: " + string(c.data)}
@@ -188,8 +215,8 @@ func findIconSlow(ctx context.Context, r httpRoute, uri string, depth int) *Fetc
 	}
 	switch {
 	case strings.HasPrefix(href, "http://"), strings.HasPrefix(href, "https://"):
-		return fetchIconAbsolute(href)
+		return fetchIconAbsolute(ctx, href)
 	default:
-		return findIconSlow(ctx, r, href, 0)
+		return findIconSlow(ctx, r, href, append(stack, newReq.URL.Path))
 	}
 }

internal/homepage/icon_cache.go

@@ -1,6 +1,7 @@
 package homepage
 
 import (
+	"encoding/base64"
 	"sync"
 	"time"
 
@@ -10,11 +11,13 @@ import (
 	"github.com/yusing/go-proxy/internal/logging"
 	"github.com/yusing/go-proxy/internal/task"
 	"github.com/yusing/go-proxy/internal/utils"
+	"github.com/yusing/go-proxy/internal/utils/atomic"
 )
 
 type cacheEntry struct {
-	Icon       []byte    `json:"icon"`
+	Icon        []byte                  `json:"icon"`
-	LastAccess time.Time `json:"lastAccess"`
+	ContentType string                  `json:"content_type"`
+	LastAccess  atomic.Value[time.Time] `json:"last_access"`
 }
 
 // cache key can be absolute url or route name.
@@ -25,7 +28,9 @@ var (
 
 const (
 	iconCacheTTL    = 3 * 24 * time.Hour
-	cleanUpInterval = time.Hour
+	cleanUpInterval = time.Minute
+	maxCacheSize    = 1024 * 1024 // 1MB
+	maxCacheEntries = 100
 )
 
 func InitIconCache() {
@@ -77,19 +82,29 @@ func pruneExpiredIconCache() {
 			nPruned++
 		}
 	}
+	if len(iconCache) > maxCacheEntries {
+		newIconCache := make(map[string]*cacheEntry, maxCacheEntries)
+		i := 0
+		for key, icon := range iconCache {
+			if i == maxCacheEntries {
+				break
+			}
+			if !icon.IsExpired() {
+				newIconCache[key] = icon
+				i++
+			}
+		}
+		iconCache = newIconCache
+	}
 	if nPruned > 0 {
 		logging.Info().Int("pruned", nPruned).Msg("pruned expired icon cache")
 	}
 }
 
-func routeKey(r route) string {
-	return r.ProviderName() + ":" + r.TargetName()
-}
-
 func PruneRouteIconCache(route route) {
 	iconCacheMu.Lock()
 	defer iconCacheMu.Unlock()
-	delete(iconCache, routeKey(route))
+	delete(iconCache, route.Key())
 }
 
 func loadIconCache(key string) *FetchResult {
@@ -97,41 +112,49 @@ func loadIconCache(key string) *FetchResult {
 	defer iconCacheMu.RUnlock()
 
 	icon, ok := iconCache[key]
-	if ok && icon != nil {
+	if ok && len(icon.Icon) > 0 {
 		logging.Debug().
 			Str("key", key).
 			Msg("icon found in cache")
-		icon.LastAccess = time.Now()
+		icon.LastAccess.Store(time.Now())
-		return &FetchResult{Icon: icon.Icon}
+		return &FetchResult{Icon: icon.Icon, contentType: icon.ContentType}
 	}
 	return nil
 }
 
-func storeIconCache(key string, icon []byte) {
+func storeIconCache(key string, result *FetchResult) {
+	icon := result.Icon
+	if len(icon) > maxCacheSize {
+		logging.Debug().Int("size", len(icon)).Msg("icon cache size exceeds max cache size")
+		return
+	}
 	iconCacheMu.Lock()
 	defer iconCacheMu.Unlock()
-	iconCache[key] = &cacheEntry{Icon: icon, LastAccess: time.Now()}
+	entry := &cacheEntry{Icon: icon, ContentType: result.contentType}
+	entry.LastAccess.Store(time.Now())
+	iconCache[key] = entry
+	logging.Debug().Str("key", key).Int("size", len(icon)).Msg("stored icon cache")
 }
 
 func (e *cacheEntry) IsExpired() bool {
-	return time.Since(e.LastAccess) > iconCacheTTL
+	return time.Since(e.LastAccess.Load()) > iconCacheTTL
 }
 
 func (e *cacheEntry) UnmarshalJSON(data []byte) error {
-	attempt := struct {
+	// check if data is json
-		Icon       []byte    `json:"icon"`
+	if json.Valid(data) {
-		LastAccess time.Time `json:"lastAccess"`
+		err := json.Unmarshal(data, &e)
-	}{}
+		// return only if unmarshal is successful
-	err := json.Unmarshal(data, &attempt)
+		// otherwise fallback to base64
-	if err == nil {
+		if err == nil {
-		e.Icon = attempt.Icon
+			return nil
-		e.LastAccess = attempt.LastAccess
+		}
-		return nil
 	}
-	// fallback to bytes
+	// fallback to base64
-	err = json.Unmarshal(data, &e.Icon)
+	icon, err := base64.StdEncoding.DecodeString(string(data))
 	if err == nil {
-		e.LastAccess = time.Now()
+		e.Icon = icon
+		e.LastAccess.Store(time.Now())
 		return nil
 	}
 	return err

internal/homepage/list-icons.go

@@ -60,15 +60,15 @@ func InitIconListCache() {
 		DisplayNames: make(ReferenceDisplayNameMap),
 		IconList:     []string{},
 	}
-	// err := utils.LoadJSONIfExist(common.IconListCachePath, iconsCache)
+	err := utils.LoadJSONIfExist(common.IconListCachePath, iconsCache)
-	// if err != nil {
+	if err != nil {
-	// 	logging.Error().Err(err).Msg("failed to load icon list cache config")
+		logging.Error().Err(err).Msg("failed to load icon list cache config")
-	// } else if len(iconsCache.IconList) > 0 {
+	} else if len(iconsCache.IconList) > 0 {
-	// 	logging.Info().
+		logging.Info().
-	// 		Int("icons", len(iconsCache.IconList)).
+			Int("icons", len(iconsCache.IconList)).
-	// 		Int("display_names", len(iconsCache.DisplayNames)).
+			Int("display_names", len(iconsCache.DisplayNames)).
-	// 		Msg("icon list cache loaded")
+			Msg("icon list cache loaded")
-	// }
+	}
 }
 
 func ListAvailableIcons() (*Cache, error) {

internal/homepage/route.go

@@ -3,10 +3,12 @@ package homepage
 import (
 	"net/http"
 	"net/url"
+
+	"github.com/yusing/go-proxy/internal/utils/pool"
 )
 
 type route interface {
-	TargetName() string
+	pool.Object
 	ProviderName() string
 	Reference() string
 	TargetURL() *url.URL

internal/idlewatcher/watcher.go

@@ -13,7 +13,7 @@ import (
 	"github.com/yusing/go-proxy/internal/logging"
 	"github.com/yusing/go-proxy/internal/net/gphttp/reverseproxy"
 	net "github.com/yusing/go-proxy/internal/net/types"
-	route "github.com/yusing/go-proxy/internal/route/types"
+	"github.com/yusing/go-proxy/internal/route/routes"
 	"github.com/yusing/go-proxy/internal/task"
 	U "github.com/yusing/go-proxy/internal/utils"
 	"github.com/yusing/go-proxy/internal/utils/atomic"
@@ -80,7 +80,7 @@ var (
 const reqTimeout = 3 * time.Second
 
 // TODO: fix stream type
-func NewWatcher(parent task.Parent, r route.Route) (*Watcher, error) {
+func NewWatcher(parent task.Parent, r routes.Route) (*Watcher, error) {
 	cfg := r.IdlewatcherConfig()
 	key := cfg.Key()
 
@@ -126,9 +126,9 @@ func NewWatcher(parent task.Parent, r route.Route) (*Watcher, error) {
 		Logger()
 
 	switch r := r.(type) {
-	case route.ReverseProxyRoute:
+	case routes.ReverseProxyRoute:
 		w.rp = r.ReverseProxy()
-	case route.StreamRoute:
+	case routes.StreamRoute:
 		w.stream = r
 	default:
 		return nil, gperr.New("unexpected route type")
@@ -153,14 +153,14 @@ func NewWatcher(parent task.Parent, r route.Route) (*Watcher, error) {
 
 	w.state.Store(&containerState{status: status})
 
-	w.task = parent.Subtask("idlewatcher."+r.TargetName(), true)
+	w.task = parent.Subtask("idlewatcher."+r.Name(), true)
 
 	watcherMapMu.Lock()
 	defer watcherMapMu.Unlock()
 	watcherMap[key] = w
 	go func() {
 		cause := w.watchUntilDestroy()
-		if cause.Is(causeContainerDestroy) {
+		if cause.Is(causeContainerDestroy) || cause.Is(task.ErrProgramExiting) {
 			watcherMapMu.Lock()
 			defer watcherMapMu.Unlock()
 			delete(watcherMap, key)
@@ -173,7 +173,11 @@ func NewWatcher(parent task.Parent, r route.Route) (*Watcher, error) {
 		w.provider.Close()
 		w.task.Finish(cause)
 	}()
-	w.l.Info().Msg("idlewatcher started")
+	if exists {
+		w.l.Info().Msg("idlewatcher reloaded")
+	} else {
+		w.l.Info().Msg("idlewatcher started")
+	}
 	return w, nil
 }
 

internal/metrics/period/poller.go

@@ -133,7 +133,7 @@ func (p *Poller[T, AggregateT]) pollWithTimeout(ctx context.Context) {
 }
 
 func (p *Poller[T, AggregateT]) Start() {
-	t := task.RootTask("poller." + p.name)
+	t := task.RootTask("poller."+p.name, true)
 	err := p.load()
 	if err != nil {
 		if !os.IsNotExist(err) {

internal/metrics/uptime/uptime.go

@@ -10,15 +10,14 @@ import (
 	"github.com/yusing/go-proxy/internal/metrics/period"
 	metricsutils "github.com/yusing/go-proxy/internal/metrics/utils"
 	"github.com/yusing/go-proxy/internal/route/routes"
-	"github.com/yusing/go-proxy/internal/route/routes/routequery"
 	"github.com/yusing/go-proxy/internal/watcher/health"
 	"github.com/yusing/go-proxy/pkg/json"
 )
 
 type (
 	StatusByAlias struct {
-		Map       json.Map[*routequery.HealthInfoRaw] `json:"statuses"`
+		Map       json.Map[*routes.HealthInfoRaw] `json:"statuses"`
-		Timestamp int64                               `json:"timestamp"`
+		Timestamp int64                           `json:"timestamp"`
 	}
 	Aggregated = json.MapSlice[any]
 )
@@ -27,7 +26,7 @@ var Poller = period.NewPoller("uptime", getStatuses, aggregateStatuses)
 
 func getStatuses(ctx context.Context, _ *StatusByAlias) (*StatusByAlias, error) {
 	return &StatusByAlias{
-		Map:       routequery.HealthInfo(),
+		Map:       routes.HealthInfo(),
 		Timestamp: time.Now().Unix(),
 	}, nil
 }
@@ -111,7 +110,7 @@ func (rs RouteStatuses) aggregate(limit int, offset int) Aggregated {
 			"avg_latency": latency,
 			"statuses":    statuses,
 		}
-		r, ok := routes.GetRoute(alias)
+		r, ok := routes.HTTP.Get(alias)
 		if ok {
 			result[i]["display_name"] = r.HomepageConfig().Name
 		}

internal/net/gphttp/accesslog/access_logger.go

@@ -11,6 +11,7 @@ import (
 	"github.com/yusing/go-proxy/internal/gperr"
 	"github.com/yusing/go-proxy/internal/logging"
 	"github.com/yusing/go-proxy/internal/task"
+	"github.com/yusing/go-proxy/internal/utils/synk"
 )
 
 type (
@@ -20,7 +21,7 @@ type (
 		io       AccessLogIO
 		buffered *bufio.Writer
 
-		lineBufPool sync.Pool // buffer pool for formatting a single log line
+		lineBufPool *synk.BytesPool // buffer pool for formatting a single log line
 		Formatter
 	}
 
@@ -78,7 +79,7 @@ func NewAccessLoggerWithIO(parent task.Parent, io AccessLogIO, cfg *Config) *Acc
 		cfg.BufferSize = 4096
 	}
 	l := &AccessLogger{
-		task:     parent.Subtask("accesslog"),
+		task:     parent.Subtask("accesslog."+io.Name(), true),
 		cfg:      cfg,
 		io:       io,
 		buffered: bufio.NewWriterSize(io, cfg.BufferSize),
@@ -96,9 +97,7 @@ func NewAccessLoggerWithIO(parent task.Parent, io AccessLogIO, cfg *Config) *Acc
 		panic("invalid access log format")
 	}
 
-	l.lineBufPool.New = func() any {
+	l.lineBufPool = synk.NewBytesPool(1024, synk.DefaultMaxBytes)
-		return bytes.NewBuffer(make([]byte, 0, 1024))
-	}
 	go l.start()
 	return l
 }
@@ -118,12 +117,11 @@ func (l *AccessLogger) Log(req *http.Request, res *http.Response) {
 		return
 	}
 
-	line := l.lineBufPool.Get().(*bytes.Buffer)
+	line := l.lineBufPool.Get()
-	line.Reset()
 	defer l.lineBufPool.Put(line)
-	l.Formatter.Format(line, req, res)
+	l.Formatter.Format(bytes.NewBuffer(line), req, res)
-	line.WriteRune('\n')
+	line = append(line, '\n')
-	l.write(line.Bytes())
+	l.write(line)
 }
 
 func (l *AccessLogger) LogError(req *http.Request, err error) {

internal/net/gphttp/accesslog/access_logger_test.go

@@ -23,7 +23,7 @@ const (
 	referer       = "https://www.google.com/"
 	proto         = "HTTP/1.1"
 	ua            = "Go-http-client/1.1"
-	status        = http.StatusOK
+	status        = http.StatusNotFound
 	contentLength = 100
 	method        = http.MethodGet
 )
@@ -99,6 +99,25 @@ func TestAccessLoggerRedactQuery(t *testing.T) {
 	)
 }
 
+type JSONLogEntry struct {
+	Time        string              `json:"time"`
+	IP          string              `json:"ip"`
+	Method      string              `json:"method"`
+	Scheme      string              `json:"scheme"`
+	Host        string              `json:"host"`
+	URI         string              `json:"uri"`
+	Protocol    string              `json:"protocol"`
+	Status      int                 `json:"status"`
+	Error       string              `json:"error,omitempty"`
+	ContentType string              `json:"type"`
+	Size        int64               `json:"size"`
+	Referer     string              `json:"referer"`
+	UserAgent   string              `json:"useragent"`
+	Query       map[string][]string `json:"query,omitempty"`
+	Headers     map[string][]string `json:"headers,omitempty"`
+	Cookies     map[string]string   `json:"cookies,omitempty"`
+}
+
 func getJSONEntry(t *testing.T, config *Config) JSONLogEntry {
 	t.Helper()
 	config.Format = FormatJSON
@@ -125,4 +144,7 @@ func TestAccessLoggerJSON(t *testing.T) {
 	ExpectEqual(t, entry.UserAgent, ua)
 	ExpectEqual(t, len(entry.Headers), 0)
 	ExpectEqual(t, len(entry.Cookies), 0)
+	if status >= 400 {
+		ExpectEqual(t, entry.Error, http.StatusText(status))
+	}
 }

internal/net/gphttp/accesslog/formatter.go

@@ -8,9 +8,8 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/rs/zerolog"
 	"github.com/yusing/go-proxy/pkg/json"
-
-	"github.com/yusing/go-proxy/internal/logging"
 )
 
 type (
@@ -20,25 +19,6 @@ type (
 	}
 	CombinedFormatter struct{ CommonFormatter }
 	JSONFormatter     struct{ CommonFormatter }
-
-	JSONLogEntry struct {
-		Time        string              `json:"time"`
-		IP          string              `json:"ip"`
-		Method      string              `json:"method"`
-		Scheme      string              `json:"scheme"`
-		Host        string              `json:"host"`
-		URI         string              `json:"uri"`
-		Protocol    string              `json:"protocol"`
-		Status      int                 `json:"status"`
-		Error       string              `json:"error,omitempty"`
-		ContentType string              `json:"type"`
-		Size        int64               `json:"size"`
-		Referer     string              `json:"referer"`
-		UserAgent   string              `json:"useragent"`
-		Query       map[string][]string `json:"query,omitempty"`
-		Headers     map[string][]string `json:"headers,omitempty"`
-		Cookies     map[string]string   `json:"cookies,omitempty"`
-	}
 )
 
 const LogTimeFormat = "02/Jan/2006:15:04:05 -0700"
@@ -109,37 +89,36 @@ func (f *JSONFormatter) Format(line *bytes.Buffer, req *http.Request, res *http.
 	headers := f.cfg.Headers.ProcessHeaders(req.Header)
 	headers.Del("Cookie")
 	cookies := f.cfg.Cookies.ProcessCookies(req.Cookies())
+	contentType := res.Header.Get("Content-Type")
 
-	entry := JSONLogEntry{
+	queryBytes, _ := json.Marshal(query)
-		Time:        f.GetTimeNow().Format(LogTimeFormat),
+	headersBytes, _ := json.Marshal(headers)
-		IP:          clientIP(req),
+	cookiesBytes, _ := json.Marshal(cookies)
-		Method:      req.Method,
+
-		Scheme:      scheme(req),
+	logger := zerolog.New(line).With().Logger()
-		Host:        req.Host,
+	event := logger.Info().
-		URI:         requestURI(req.URL, query),
+		Str("time", f.GetTimeNow().Format(LogTimeFormat)).
-		Protocol:    req.Proto,
+		Str("ip", clientIP(req)).
-		Status:      res.StatusCode,
+		Str("method", req.Method).
-		ContentType: res.Header.Get("Content-Type"),
+		Str("scheme", scheme(req)).
-		Size:        res.ContentLength,
+		Str("host", req.Host).
-		Referer:     req.Referer(),
+		Str("uri", requestURI(req.URL, query)).
-		UserAgent:   req.UserAgent(),
+		Str("protocol", req.Proto).
-		Query:       query,
+		Int("status", res.StatusCode).
-		Headers:     headers,
+		Str("type", contentType).
-		Cookies:     cookies,
+		Int64("size", res.ContentLength).
-	}
+		Str("referer", req.Referer()).
+		Str("useragent", req.UserAgent()).
+		RawJSON("query", queryBytes).
+		RawJSON("headers", headersBytes).
+		RawJSON("cookies", cookiesBytes)
 
 	if res.StatusCode >= 400 {
-		entry.Error = res.Status
+		if res.Status != "" {
-	}
+			event.Str("error", res.Status)
-
+		} else {
-	if entry.ContentType == "" {
+			event.Str("error", http.StatusText(res.StatusCode))
-		// try to get content type from request
+		}
-		entry.ContentType = req.Header.Get("Content-Type")
-	}
-
-	marshaller := json.NewEncoder(line)
-	err := marshaller.Encode(entry)
-	if err != nil {
-		logging.Err(err).Msg("failed to marshal json log")
 	}
+	event.Send()
 }

internal/net/gphttp/default_client.go

@@ -24,4 +24,5 @@ var (
 	Get  = httpClient.Get
 	Post = httpClient.Post
 	Head = httpClient.Head
+	Do   = httpClient.Do
 )

internal/net/gphttp/loadbalancer/loadbalancer.go

@@ -1,6 +1,7 @@
 package loadbalancer
 
 import (
+	"fmt"
 	"net/http"
 	"sync"
 	"time"
@@ -10,8 +11,8 @@ import (
 	"github.com/yusing/go-proxy/internal/logging"
 	"github.com/yusing/go-proxy/internal/net/gphttp/httpheaders"
 	"github.com/yusing/go-proxy/internal/net/gphttp/loadbalancer/types"
-	"github.com/yusing/go-proxy/internal/route/routes"
 	"github.com/yusing/go-proxy/internal/task"
+	"github.com/yusing/go-proxy/internal/utils/pool"
 	"github.com/yusing/go-proxy/internal/watcher/health"
 )
 
@@ -30,7 +31,7 @@ type (
 
 		task *task.Task
 
-		pool   Pool
+		pool   pool.Pool[Server]
 		poolMu sync.Mutex
 
 		sumWeight Weight
@@ -45,7 +46,7 @@ const maxWeight Weight = 100
 func New(cfg *Config) *LoadBalancer {
 	lb := &LoadBalancer{
 		Config: new(Config),
-		pool:   types.NewServerPool(),
+		pool:   pool.New[Server]("loadbalancer." + cfg.Link),
 		l:      logging.With().Str("name", cfg.Link).Logger(),
 	}
 	lb.UpdateConfigIfNeeded(cfg)
@@ -55,16 +56,14 @@ func New(cfg *Config) *LoadBalancer {
 // Start implements task.TaskStarter.
 func (lb *LoadBalancer) Start(parent task.Parent) gperr.Error {
 	lb.startTime = time.Now()
-	lb.task = parent.Subtask("loadbalancer."+lb.Link, false)
+	lb.task = parent.Subtask("loadbalancer."+lb.Link, true)
-	parent.OnCancel("lb_remove_route", func() {
+	lb.task.OnCancel("cleanup", func() {
-		routes.DeleteHTTPRoute(lb.Link)
-	})
-	lb.task.OnFinished("cleanup", func() {
 		if lb.impl != nil {
-			lb.pool.RangeAll(func(k string, v Server) {
+			for _, srv := range lb.pool.Iter {
-				lb.impl.OnRemoveServer(v)
+				lb.impl.OnRemoveServer(srv)
-			})
+			}
 		}
+		lb.task.Finish(nil)
 	})
 	return nil
 }
@@ -90,9 +89,9 @@ func (lb *LoadBalancer) updateImpl() {
 	default: // should happen in test only
 		lb.impl = lb.newRoundRobin()
 	}
-	lb.pool.RangeAll(func(_ string, srv Server) {
+	for _, srv := range lb.pool.Iter {
 		lb.impl.OnAddServer(srv)
-	})
+	}
 }
 
 func (lb *LoadBalancer) UpdateConfigIfNeeded(cfg *Config) {
@@ -124,12 +123,12 @@ func (lb *LoadBalancer) AddServer(srv Server) {
 	lb.poolMu.Lock()
 	defer lb.poolMu.Unlock()
 
-	if lb.pool.Has(srv.Key()) { // FIXME: this should be a warning
+	if old, ok := lb.pool.Get(srv.Key()); ok { // FIXME: this should be a warning
-		old, _ := lb.pool.Load(srv.Key())
 		lb.sumWeight -= old.Weight()
 		lb.impl.OnRemoveServer(old)
+		lb.pool.Del(old)
 	}
-	lb.pool.Store(srv.Key(), srv)
+	lb.pool.Add(srv)
 	lb.sumWeight += srv.Weight()
 
 	lb.rebalance()
@@ -145,11 +144,11 @@ func (lb *LoadBalancer) RemoveServer(srv Server) {
 	lb.poolMu.Lock()
 	defer lb.poolMu.Unlock()
 
-	if !lb.pool.Has(srv.Key()) {
+	if _, ok := lb.pool.Get(srv.Key()); !ok {
 		return
 	}
 
-	lb.pool.Delete(srv.Key())
+	lb.pool.Del(srv)
 
 	lb.sumWeight -= srv.Weight()
 	lb.rebalance()
@@ -178,15 +177,15 @@ func (lb *LoadBalancer) rebalance() {
 	if lb.sumWeight == 0 { // distribute evenly
 		weightEach := maxWeight / Weight(poolSize)
 		remainder := maxWeight % Weight(poolSize)
-		lb.pool.RangeAll(func(_ string, s Server) {
+		for _, srv := range lb.pool.Iter {
 			w := weightEach
 			lb.sumWeight += weightEach
 			if remainder > 0 {
 				w++
 				remainder--
 			}
-			s.SetWeight(w)
+			srv.SetWeight(w)
-		})
+		}
 		return
 	}
 
@@ -194,30 +193,29 @@ func (lb *LoadBalancer) rebalance() {
 	scaleFactor := float64(maxWeight) / float64(lb.sumWeight)
 	lb.sumWeight = 0
 
-	lb.pool.RangeAll(func(_ string, s Server) {
+	for _, srv := range lb.pool.Iter {
-		s.SetWeight(Weight(float64(s.Weight()) * scaleFactor))
+		srv.SetWeight(Weight(float64(srv.Weight()) * scaleFactor))
-		lb.sumWeight += s.Weight()
+		lb.sumWeight += srv.Weight()
-	})
+	}
 
 	delta := maxWeight - lb.sumWeight
 	if delta == 0 {
 		return
 	}
-	lb.pool.Range(func(_ string, s Server) bool {
+	for _, srv := range lb.pool.Iter {
 		if delta == 0 {
-			return false
+			break
 		}
 		if delta > 0 {
-			s.SetWeight(s.Weight() + 1)
+			srv.SetWeight(srv.Weight() + 1)
 			lb.sumWeight++
 			delta--
 		} else {
-			s.SetWeight(s.Weight() - 1)
+			srv.SetWeight(srv.Weight() - 1)
 			lb.sumWeight--
 			delta++
 		}
-		return true
+	}
-	})
 }
 
 func (lb *LoadBalancer) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
@@ -242,13 +240,16 @@ func (lb *LoadBalancer) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 // MarshalMap implements health.HealthMonitor.
 func (lb *LoadBalancer) MarshalMap() map[string]any {
 	extra := make(map[string]any)
-	lb.pool.RangeAll(func(k string, v Server) {
+	for _, srv := range lb.pool.Iter {
-		extra[v.Key()] = v
+		extra[srv.Key()] = srv
-	})
+	}
+
+	status, numHealthy := lb.status()
 
 	return (&health.JSONRepresentation{
 		Name:    lb.Name(),
-		Status:  lb.Status(),
+		Status:  status,
+		Detail:  fmt.Sprintf("%d/%d servers are healthy", numHealthy, lb.pool.Size()),
 		Started: lb.startTime,
 		Uptime:  lb.Uptime(),
 		Extra: map[string]any{
@@ -265,22 +266,26 @@ func (lb *LoadBalancer) Name() string {
 
 // Status implements health.HealthMonitor.
 func (lb *LoadBalancer) Status() health.Status {
+	status, _ := lb.status()
+	return status
+}
+
+func (lb *LoadBalancer) status() (status health.Status, numHealthy int) {
 	if lb.pool.Size() == 0 {
-		return health.StatusUnknown
+		return health.StatusUnknown, 0
 	}
 
-	isHealthy := true
+	// should be healthy if at least one server is healthy
-	lb.pool.Range(func(_ string, srv Server) bool {
+	numHealthy = 0
-		if srv.Status().Bad() {
+	for _, srv := range lb.pool.Iter {
-			isHealthy = false
+		if srv.Status().Good() {
-			return false
+			numHealthy++
 		}
-		return true
-	})
-	if !isHealthy {
-		return health.StatusUnhealthy
 	}
-	return health.StatusHealthy
+	if numHealthy == 0 {
+		return health.StatusUnhealthy, numHealthy
+	}
+	return health.StatusHealthy, numHealthy
 }
 
 // Uptime implements health.HealthMonitor.
@@ -291,9 +296,9 @@ func (lb *LoadBalancer) Uptime() time.Duration {
 // Latency implements health.HealthMonitor.
 func (lb *LoadBalancer) Latency() time.Duration {
 	var sum time.Duration
-	lb.pool.RangeAll(func(_ string, srv Server) {
+	for _, srv := range lb.pool.Iter {
 		sum += srv.Latency()
-	})
+	}
 	return sum
 }
 
@@ -304,10 +309,10 @@ func (lb *LoadBalancer) String() string {
 
 func (lb *LoadBalancer) availServers() []Server {
 	avail := make([]Server, 0, lb.pool.Size())
-	lb.pool.RangeAll(func(_ string, srv Server) {
+	for _, srv := range lb.pool.Iter {
 		if srv.Status().Good() {
 			avail = append(avail, srv)
 		}
-	})
+	}
 	return avail
 }

internal/net/gphttp/loadbalancer/types.go

@@ -7,7 +7,6 @@ import (
 type (
 	Server  = types.Server
 	Servers = []types.Server
-	Pool    = types.Pool
 	Weight  = types.Weight
 	Config  = types.Config
 	Mode    = types.Mode

internal/net/gphttp/loadbalancer/types/server.go

@@ -6,7 +6,6 @@ import (
 
 	idlewatcher "github.com/yusing/go-proxy/internal/idlewatcher/types"
 	U "github.com/yusing/go-proxy/internal/utils"
-	F "github.com/yusing/go-proxy/internal/utils/functional"
 	"github.com/yusing/go-proxy/internal/watcher/health"
 )
 
@@ -32,12 +31,8 @@ type (
 		SetWeight(weight Weight)
 		TryWake() error
 	}
-
-	Pool = F.Map[string, Server]
 )
 
-var NewServerPool = F.NewMap[Pool]
-
 func NewServer(name string, url *url.URL, weight Weight, handler http.Handler, healthMon health.HealthMonitor) Server {
 	srv := &server{
 		name:          name,

internal/net/gphttp/server/server.go

@@ -96,7 +96,9 @@ func (s *Server) Start(parent task.Parent) {
 			TLSConfig: http3.ConfigureTLSConfig(s.https.TLSConfig),
 		}
 		Start(subtask, h3, &s.l)
-		s.http.Handler = advertiseHTTP3(s.http.Handler, h3)
+		if s.http != nil {
+			s.http.Handler = advertiseHTTP3(s.http.Handler, h3)
+		}
 		s.https.Handler = advertiseHTTP3(s.https.Handler, h3)
 	}
 

internal/net/ping.go

@@ -94,24 +94,20 @@ func Ping(ctx context.Context, ip net.IP) (bool, error) {
 	}
 }
 
-var pingDialer = &net.Dialer{
-	Timeout: 2 * time.Second,
-}
-
 // PingWithTCPFallback pings the IP address using ICMP and TCP fallback.
 //
 // If the ICMP ping fails due to permission error, it will try to connect to the specified port.
 func PingWithTCPFallback(ctx context.Context, ip net.IP, port int) (bool, error) {
 	ok, err := Ping(ctx, ip)
-	if err != nil {
+	if err == nil {
-		if !errors.Is(err, os.ErrPermission) {
-			return false, err
-		}
-	} else {
 		return ok, nil
 	}
+	if !errors.Is(err, os.ErrPermission) {
+		return false, err
+	}
 
-	conn, err := pingDialer.DialContext(ctx, "tcp", fmt.Sprintf("%s:%d", ip, port))
+	var dialer net.Dialer
+	conn, err := dialer.DialContext(ctx, "tcp", fmt.Sprintf("%s:%d", ip, port))
 	if err != nil {
 		return false, err
 	}

internal/net/types/stream.go

@@ -1,13 +1,11 @@
 package gpnet
 
 import (
-	"fmt"
 	"net"
 )
 
 type (
 	Stream interface {
-		fmt.Stringer
 		StreamListener
 		Setup() error
 		Handle(conn StreamConn) error

internal/notif/dispatcher.go

@@ -33,7 +33,7 @@ const dispatchErr = "notification dispatch error"
 
 func StartNotifDispatcher(parent task.Parent) *Dispatcher {
 	dispatcher = &Dispatcher{
-		task:      parent.Subtask("notification"),
+		task:      parent.Subtask("notification", true),
 		logCh:     make(chan *LogMessage),
 		providers: F.NewSet[Provider](),
 	}
@@ -86,7 +86,7 @@ func (disp *Dispatcher) dispatch(msg *LogMessage) {
 	if true {
 		return
 	}
-	task := disp.task.Subtask("dispatcher")
+	task := disp.task.Subtask("dispatcher", true)
 	defer task.Finish("notif dispatched")
 
 	errs := gperr.NewBuilder(dispatchErr)

internal/route/fileserver.go

@@ -57,7 +57,7 @@ func NewFileServer(base *Route) (*FileServer, gperr.Error) {
 
 // Start implements task.TaskStarter.
 func (s *FileServer) Start(parent task.Parent) gperr.Error {
-	s.task = parent.Subtask("fileserver."+s.TargetName(), false)
+	s.task = parent.Subtask("fileserver."+s.Name(), false)
 
 	pathPatterns := s.PathPatterns
 	switch {
@@ -92,7 +92,7 @@ func (s *FileServer) Start(parent task.Parent) gperr.Error {
 	}
 
 	if common.PrometheusEnabled {
-		metricsLogger := metricslogger.NewMetricsLogger(s.TargetName())
+		metricsLogger := metricslogger.NewMetricsLogger(s.Name())
 		s.handler = metricsLogger.GetHandler(s.handler)
 		s.task.OnCancel("reset_metrics", metricsLogger.ResetMetrics)
 	}
@@ -104,9 +104,9 @@ func (s *FileServer) Start(parent task.Parent) gperr.Error {
 		}
 	}
 
-	routes.SetHTTPRoute(s.TargetName(), s)
+	routes.HTTP.Add(s)
 	s.task.OnCancel("entrypoint_remove_route", func() {
-		routes.DeleteHTTPRoute(s.TargetName())
+		routes.HTTP.Del(s)
 	})
 	return nil
 }

internal/route/reverse_proxy.go

@@ -59,7 +59,7 @@ func NewReverseProxyRoute(base *Route) (*ReveseProxyRoute, gperr.Error) {
 		}
 	}
 
-	service := base.TargetName()
+	service := base.Name()
 	rp := reverseproxy.NewReverseProxy(service, proxyURL, trans)
 
 	if len(base.Middlewares) > 0 {
@@ -90,16 +90,12 @@ func NewReverseProxyRoute(base *Route) (*ReveseProxyRoute, gperr.Error) {
 	return r, nil
 }
 
-func (r *ReveseProxyRoute) String() string {
-	return r.TargetName()
-}
-
 // Start implements task.TaskStarter.
 func (r *ReveseProxyRoute) Start(parent task.Parent) gperr.Error {
-	if existing, ok := routes.GetHTTPRoute(r.TargetName()); ok && !r.UseLoadBalance() {
+	if existing, ok := routes.HTTP.Get(r.Key()); ok && !r.UseLoadBalance() {
 		return gperr.Errorf("route already exists: from provider %s and %s", existing.ProviderName(), r.ProviderName())
 	}
-	r.task = parent.Subtask("http."+r.TargetName(), false)
+	r.task = parent.Subtask("http."+r.Name(), false)
 
 	switch {
 	case r.UseIdleWatcher():
@@ -132,7 +128,7 @@ func (r *ReveseProxyRoute) Start(parent task.Parent) gperr.Error {
 			r.handler = r.rp
 		default:
 			logging.Warn().
-				Str("route", r.TargetName()).
+				Str("route", r.Name()).
 				Msg("`path_patterns` for reverse proxy is deprecated. Use `rules` instead.")
 			mux := gphttp.NewServeMux()
 			patErrs := gperr.NewBuilder("invalid path pattern(s)")
@@ -148,7 +144,7 @@ func (r *ReveseProxyRoute) Start(parent task.Parent) gperr.Error {
 	}
 
 	if len(r.Rules) > 0 {
-		r.handler = r.Rules.BuildHandler(r.TargetName(), r.handler)
+		r.handler = r.Rules.BuildHandler(r.Name(), r.handler)
 	}
 
 	if r.HealthMon != nil {
@@ -158,7 +154,7 @@ func (r *ReveseProxyRoute) Start(parent task.Parent) gperr.Error {
 	}
 
 	if common.PrometheusEnabled {
-		metricsLogger := metricslogger.NewMetricsLogger(r.TargetName())
+		metricsLogger := metricslogger.NewMetricsLogger(r.Name())
 		r.handler = metricsLogger.GetHandler(r.handler)
 		r.task.OnCancel("reset_metrics", metricsLogger.ResetMetrics)
 	}
@@ -166,9 +162,9 @@ func (r *ReveseProxyRoute) Start(parent task.Parent) gperr.Error {
 	if r.UseLoadBalance() {
 		r.addToLoadBalancer(parent)
 	} else {
-		routes.SetHTTPRoute(r.TargetName(), r)
+		routes.HTTP.Add(r)
 		r.task.OnFinished("entrypoint_remove_route", func() {
-			routes.DeleteHTTPRoute(r.TargetName())
+			routes.HTTP.Del(r)
 		})
 	}
 
@@ -201,7 +197,7 @@ func (r *ReveseProxyRoute) HealthMonitor() health.HealthMonitor {
 func (r *ReveseProxyRoute) addToLoadBalancer(parent task.Parent) {
 	var lb *loadbalancer.LoadBalancer
 	cfg := r.LoadBalance
-	l, ok := routes.GetHTTPRoute(cfg.Link)
+	l, ok := routes.HTTP.Get(cfg.Link)
 	var linked *ReveseProxyRoute
 	if ok {
 		linked = l.(*ReveseProxyRoute)
@@ -222,7 +218,10 @@ func (r *ReveseProxyRoute) addToLoadBalancer(parent task.Parent) {
 			loadBalancer: lb,
 			handler:      lb,
 		}
-		routes.SetHTTPRoute(cfg.Link, linked)
+		routes.HTTP.Add(linked)
+		r.task.OnFinished("entrypoint_remove_route", func() {
+			routes.HTTP.Del(linked)
+		})
 	}
 	r.loadBalancer = lb
 

internal/route/route.go

@@ -24,6 +24,7 @@ import (
 	config "github.com/yusing/go-proxy/internal/config/types"
 	"github.com/yusing/go-proxy/internal/net/gphttp/accesslog"
 	loadbalance "github.com/yusing/go-proxy/internal/net/gphttp/loadbalancer/types"
+	"github.com/yusing/go-proxy/internal/route/routes"
 	"github.com/yusing/go-proxy/internal/route/rules"
 	route "github.com/yusing/go-proxy/internal/route/types"
 	"github.com/yusing/go-proxy/internal/utils"
@@ -62,7 +63,7 @@ type (
 		LisURL   *url.URL `json:"lurl,omitempty"`
 		ProxyURL *url.URL `json:"purl,omitempty"`
 
-		impl route.Route
+		impl routes.Route
 	}
 	Routes map[string]*Route
 )
@@ -77,17 +78,6 @@ func (r Routes) Contains(alias string) bool {
 func (r *Route) Validate() (err gperr.Error) {
 	r.Finalize()
 
-	// return error if route is localhost:<godoxy_port>
-	switch r.Host {
-	case "localhost", "127.0.0.1":
-		switch r.Port.Proxy {
-		case common.ProxyHTTPPort, common.ProxyHTTPSPort, common.APIHTTPPort:
-			if r.Scheme.IsReverseProxy() || r.Scheme == route.SchemeTCP {
-				return gperr.Errorf("localhost:%d is reserved for godoxy", r.Port.Proxy)
-			}
-		}
-	}
-
 	if r.Idlewatcher != nil && r.Idlewatcher.Proxmox != nil {
 		node := r.Idlewatcher.Proxmox.Node
 		vmid := r.Idlewatcher.Proxmox.VMID
@@ -152,6 +142,17 @@ func (r *Route) Validate() (err gperr.Error) {
 		}
 	}
 
+	// return error if route is localhost:<godoxy_port>
+	switch r.Host {
+	case "localhost", "127.0.0.1":
+		switch r.Port.Proxy {
+		case common.ProxyHTTPPort, common.ProxyHTTPSPort, common.APIHTTPPort:
+			if r.Scheme.IsReverseProxy() || r.Scheme == route.SchemeTCP {
+				return gperr.Errorf("localhost:%d is reserved for godoxy", r.Port.Proxy)
+			}
+		}
+	}
+
 	errs := gperr.NewBuilder("entry validation failed")
 
 	if r.Scheme == route.SchemeFileServer {
@@ -227,7 +228,17 @@ func (r *Route) ProviderName() string {
 	return r.Provider
 }
 
-func (r *Route) TargetName() string {
+// Name implements pool.Object.
+func (r *Route) Name() string {
+	return r.Alias
+}
+
+// Key implements pool.Object.
+func (r *Route) Key() string {
+	return r.Alias
+}
+
+func (r *Route) String() string {
 	return r.Alias
 }
 

internal/route/routes/query.go

@@ -1,15 +1,14 @@
-package routequery
+package routes
 
 import (
 	"time"
 
 	"github.com/yusing/go-proxy/internal/homepage"
-	"github.com/yusing/go-proxy/internal/route/routes"
 	route "github.com/yusing/go-proxy/internal/route/types"
 	"github.com/yusing/go-proxy/internal/watcher/health"
 )
 
-func getHealthInfo(r route.Route) map[string]string {
+func getHealthInfo(r Route) map[string]string {
 	mon := r.HealthMonitor()
 	if mon == nil {
 		return map[string]string{
@@ -30,7 +29,7 @@ type HealthInfoRaw struct {
 	Latency time.Duration `json:"latency"`
 }
 
-func getHealthInfoRaw(r route.Route) *HealthInfoRaw {
+func getHealthInfoRaw(r Route) *HealthInfoRaw {
 	mon := r.HealthMonitor()
 	if mon == nil {
 		return &HealthInfoRaw{
@@ -45,69 +44,69 @@ func getHealthInfoRaw(r route.Route) *HealthInfoRaw {
 }
 
 func HealthMap() map[string]map[string]string {
-	healthMap := make(map[string]map[string]string, routes.NumRoutes())
+	healthMap := make(map[string]map[string]string, NumRoutes())
-	routes.RangeRoutes(func(alias string, r route.Route) {
+	for alias, r := range Iter {
 		healthMap[alias] = getHealthInfo(r)
-	})
+	}
 	return healthMap
 }
 
 func HealthInfo() map[string]*HealthInfoRaw {
-	healthMap := make(map[string]*HealthInfoRaw, routes.NumRoutes())
+	healthMap := make(map[string]*HealthInfoRaw, NumRoutes())
-	routes.RangeRoutes(func(alias string, r route.Route) {
+	for alias, r := range Iter {
 		healthMap[alias] = getHealthInfoRaw(r)
-	})
+	}
 	return healthMap
 }
 
 func HomepageCategories() []string {
 	check := make(map[string]struct{})
 	categories := make([]string, 0)
-	routes.GetHTTPRoutes().RangeAll(func(alias string, r route.HTTPRoute) {
+	for _, r := range HTTP.Iter {
 		item := r.HomepageConfig()
 		if item == nil || item.Category == "" {
-			return
+			continue
 		}
 		if _, ok := check[item.Category]; ok {
-			return
+			continue
 		}
 		check[item.Category] = struct{}{}
 		categories = append(categories, item.Category)
-	})
+	}
 	return categories
 }
 
 func HomepageConfig(categoryFilter, providerFilter string) homepage.Homepage {
 	hp := make(homepage.Homepage)
 
-	routes.GetHTTPRoutes().RangeAll(func(alias string, r route.HTTPRoute) {
+	for _, r := range HTTP.Iter {
 		if providerFilter != "" && r.ProviderName() != providerFilter {
-			return
+			continue
 		}
 		item := r.HomepageItem()
 		if categoryFilter != "" && item.Category != categoryFilter {
-			return
+			continue
 		}
 		hp.Add(item)
-	})
+	}
 	return hp
 }
 
-func RoutesByAlias(typeFilter ...route.RouteType) map[string]route.Route {
+func ByAlias(typeFilter ...route.RouteType) map[string]Route {
-	rts := make(map[string]route.Route)
+	rts := make(map[string]Route)
 	if len(typeFilter) == 0 || typeFilter[0] == "" {
 		typeFilter = []route.RouteType{route.RouteTypeHTTP, route.RouteTypeStream}
 	}
 	for _, t := range typeFilter {
 		switch t {
 		case route.RouteTypeHTTP:
-			routes.GetHTTPRoutes().RangeAll(func(alias string, r route.HTTPRoute) {
+			for alias, r := range HTTP.Iter {
 				rts[alias] = r
-			})
+			}
 		case route.RouteTypeStream:
-			routes.GetStreamRoutes().RangeAll(func(alias string, r route.StreamRoute) {
+			for alias, r := range Stream.Iter {
 				rts[alias] = r
-			})
+			}
 		}
 	}
 	return rts

internal/route/routes/route.go

@@ -1,4 +1,4 @@
-package route
+package routes
 
 import (
 	"net/http"
@@ -10,6 +10,7 @@ import (
 	idlewatcher "github.com/yusing/go-proxy/internal/idlewatcher/types"
 	net "github.com/yusing/go-proxy/internal/net/types"
 	"github.com/yusing/go-proxy/internal/task"
+	"github.com/yusing/go-proxy/internal/utils/pool"
 	"github.com/yusing/go-proxy/internal/watcher/health"
 
 	loadbalance "github.com/yusing/go-proxy/internal/net/gphttp/loadbalancer/types"
@@ -21,8 +22,8 @@ type (
 	Route interface {
 		task.TaskStarter
 		task.TaskFinisher
+		pool.Object
 		ProviderName() string
-		TargetName() string
 		TargetURL() *url.URL
 		HealthMonitor() health.HealthMonitor
 		Reference() string

internal/route/routes/routes.go

@@ -1,78 +1,49 @@
 package routes
 
 import (
-	route "github.com/yusing/go-proxy/internal/route/types"
+	"github.com/yusing/go-proxy/internal/utils/pool"
-	F "github.com/yusing/go-proxy/internal/utils/functional"
 )
 
 var (
-	httpRoutes   = F.NewMapOf[string, route.HTTPRoute]()
+	HTTP   = pool.New[HTTPRoute]("http_routes")
-	streamRoutes = F.NewMapOf[string, route.StreamRoute]()
+	Stream = pool.New[StreamRoute]("stream_routes")
 )
 
-func RangeRoutes(callback func(alias string, r route.Route)) {
+func Iter(yield func(alias string, r Route) bool) {
-	httpRoutes.RangeAll(func(alias string, r route.HTTPRoute) {
+	for k, r := range HTTP.Iter {
-		callback(alias, r)
+		if !yield(k, r) {
-	})
+			break
-	streamRoutes.RangeAll(func(alias string, r route.StreamRoute) {
+		}
-		callback(alias, r)
+	}
-	})
+	for k, r := range Stream.Iter {
+		if !yield(k, r) {
+			break
+		}
+	}
 }
 
 func NumRoutes() int {
-	return httpRoutes.Size() + streamRoutes.Size()
+	return HTTP.Size() + Stream.Size()
 }
 
-func GetHTTPRoutes() F.Map[string, route.HTTPRoute] {
+func Clear() {
-	return httpRoutes
+	HTTP.Clear()
+	Stream.Clear()
 }
 
-func GetStreamRoutes() F.Map[string, route.StreamRoute] {
+func GetHTTPRouteOrExact(alias, host string) (HTTPRoute, bool) {
-	return streamRoutes
+	r, ok := HTTP.Get(alias)
-}
-
-func GetHTTPRouteOrExact(alias, host string) (route.HTTPRoute, bool) {
-	r, ok := httpRoutes.Load(alias)
 	if ok {
 		return r, true
 	}
 	// try find with exact match
-	return httpRoutes.Load(host)
+	return HTTP.Get(host)
 }
 
-func GetHTTPRoute(alias string) (route.HTTPRoute, bool) {
+func Get(alias string) (Route, bool) {
-	return httpRoutes.Load(alias)
+	r, ok := HTTP.Get(alias)
-}
-
-func GetStreamRoute(alias string) (route.StreamRoute, bool) {
-	return streamRoutes.Load(alias)
-}
-
-func GetRoute(alias string) (route.Route, bool) {
-	r, ok := httpRoutes.Load(alias)
 	if ok {
 		return r, true
 	}
-	return streamRoutes.Load(alias)
+	return Stream.Get(alias)
-}
-
-func SetHTTPRoute(alias string, r route.HTTPRoute) {
-	httpRoutes.Store(alias, r)
-}
-
-func SetStreamRoute(alias string, r route.StreamRoute) {
-	streamRoutes.Store(alias, r)
-}
-
-func DeleteHTTPRoute(alias string) {
-	httpRoutes.Delete(alias)
-}
-
-func DeleteStreamRoute(alias string) {
-	streamRoutes.Delete(alias)
-}
-
-func TestClear() {
-	httpRoutes = F.NewMapOf[string, route.HTTPRoute]()
-	streamRoutes = F.NewMapOf[string, route.StreamRoute]()
 }

internal/route/stream.go

@@ -10,7 +10,6 @@ import (
 	"github.com/yusing/go-proxy/internal/logging"
 	net "github.com/yusing/go-proxy/internal/net/types"
 	"github.com/yusing/go-proxy/internal/route/routes"
-	route "github.com/yusing/go-proxy/internal/route/types"
 	"github.com/yusing/go-proxy/internal/task"
 	"github.com/yusing/go-proxy/internal/watcher/health"
 	"github.com/yusing/go-proxy/internal/watcher/health/monitor"
@@ -29,31 +28,24 @@ type StreamRoute struct {
 	l zerolog.Logger
 }
 
-func NewStreamRoute(base *Route) (route.Route, gperr.Error) {
+func NewStreamRoute(base *Route) (routes.Route, gperr.Error) {
 	// TODO: support non-coherent scheme
 	return &StreamRoute{
 		Route: base,
 		l: logging.With().
 			Str("type", string(base.Scheme)).
-			Str("name", base.TargetName()).
+			Str("name", base.Name()).
 			Logger(),
 	}, nil
 }
 
-func (r *StreamRoute) String() string {
-	return "stream " + r.TargetName()
-}
-
 // Start implements task.TaskStarter.
 func (r *StreamRoute) Start(parent task.Parent) gperr.Error {
-	if existing, ok := routes.GetStreamRoute(r.TargetName()); ok {
+	if existing, ok := routes.Stream.Get(r.Key()); ok {
 		return gperr.Errorf("route already exists: from provider %s and %s", existing.ProviderName(), r.ProviderName())
 	}
-	r.task = parent.Subtask("stream." + r.TargetName())
+	r.task = parent.Subtask("stream."+r.Name(), true)
 	r.Stream = NewStream(r)
-	parent.OnCancel("finish", func() {
-		r.task.Finish(nil)
-	})
 
 	switch {
 	case r.UseIdleWatcher():
@@ -83,9 +75,9 @@ func (r *StreamRoute) Start(parent task.Parent) gperr.Error {
 
 	go r.acceptConnections()
 
-	routes.SetStreamRoute(r.TargetName(), r)
+	routes.Stream.Add(r)
 	r.task.OnFinished("entrypoint_remove_route", func() {
-		routes.DeleteStreamRoute(r.TargetName())
+		routes.Stream.Del(r)
 	})
 	return nil
 }

internal/task/debug.go

@@ -57,7 +57,7 @@ func (t *Task) callbackList() []map[string]any {
 func (t *Task) MarshalMap() map[string]any {
 	return map[string]any{
 		"name":          t.name,
-		"need_finish":   strconv.FormatBool(t.needFinish),
+		"need_finish":   strconv.FormatBool(t.waitFinish),
 		"childrens":     t.children,
 		"callbacks":     t.callbackList(),
 		"finish_called": t.finishedCalled,

internal/task/task.go

@@ -45,7 +45,7 @@ type (
 		callbacks     map[*Callback]struct{}
 		callbacksDone chan struct{}
 
-		needFinish bool
+		waitFinish bool
 		finished   chan struct{}
 		// finishedCalled == 1 Finish has been called
 		// but does not mean that the task is finished yet
@@ -59,7 +59,7 @@ type (
 	}
 	Parent interface {
 		Context() context.Context
-		Subtask(name string, needFinish ...bool) *Task
+		Subtask(name string, waitFinish bool) *Task
 		Name() string
 		Finish(reason any)
 		OnCancel(name string, f func())
@@ -141,13 +141,11 @@ func (t *Task) finish(reason any) {
 // Subtask returns a new subtask with the given name, derived from the parent's context.
 //
 // This should not be called after Finish is called.
-func (t *Task) Subtask(name string, needFinish ...bool) *Task {
+func (t *Task) Subtask(name string, waitFinish bool) *Task {
-	nf := len(needFinish) == 0 || needFinish[0]
-
 	ctx, cancel := context.WithCancelCause(t.ctx)
 	child := &Task{
 		parent:     t,
-		needFinish: nf,
+		waitFinish: waitFinish,
 		finished:   make(chan struct{}),
 		ctx:        ctx,
 		cancel:     cancel,
@@ -161,7 +159,7 @@ func (t *Task) Subtask(name string, needFinish ...bool) *Task {
 	allTasks.Add(child)
 	t.addChildCount()
 
-	if !nf {
+	if !waitFinish {
 		go func() {
 			<-child.ctx.Done()
 			child.Finish(nil)

internal/task/task_test.go

@@ -17,7 +17,7 @@ func TestChildTaskCancellation(t *testing.T) {
 	t.Cleanup(testCleanup)
 
 	parent := testTask()
-	child := parent.Subtask("")
+	child := parent.Subtask("child", false)
 
 	go func() {
 		defer child.Finish(nil)

internal/task/utils.go

@@ -25,8 +25,8 @@ func testCleanup() {
 }
 
 // RootTask returns a new Task with the given name, derived from the root context.
-func RootTask(name string, needFinish ...bool) *Task {
+func RootTask(name string, needFinish bool) *Task {
-	return root.Subtask(name, needFinish...)
+	return root.Subtask(name, needFinish)
 }
 
 func newRoot() *Task {
@@ -66,6 +66,9 @@ func GracefulShutdown(timeout time.Duration) (err error) {
 			return
 		case <-after:
 			logging.Warn().Msgf("Timeout waiting for %d tasks to finish", allTasks.Size())
+			for t := range allTasks.Range {
+				logging.Warn().Msgf("Task %s is still running", t.name)
+			}
 			return context.DeadlineExceeded
 		}
 	}

internal/utils/io.go

@@ -9,6 +9,7 @@ import (
 	"syscall"
 
 	"github.com/yusing/go-proxy/internal/gperr"
+	"github.com/yusing/go-proxy/internal/utils/synk"
 )
 
 // TODO: move to "utils/io".
@@ -117,24 +118,21 @@ func getHttpFlusher(dst io.Writer) httpFlusher {
 	return nil
 }
 
-const (
+const copyBufSize = 32 * 1024
-	copyBufSize = 32 * 1024
-)
 
-var copyBufPool = sync.Pool{
+var copyBufPool = synk.NewBytesPool(copyBufSize, synk.DefaultMaxBytes)
-	New: func() any {
-		return make([]byte, copyBufSize)
-	},
-}
 
 // Copyright 2009 The Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // This is a copy of io.Copy with context and HTTP flusher handling
 // Author: yusing <yusing@6uo.me>.
 func CopyClose(dst *ContextWriter, src *ContextReader) (err error) {
-	var buf []byte
+	buf := copyBufPool.Get()
+	defer copyBufPool.Put(buf)
+
+	var size int
 	if l, ok := src.Reader.(*io.LimitedReader); ok {
-		size := copyBufSize
+		size = copyBufSize
 		if int64(size) > l.N {
 			if l.N < 1 {
 				size = 1
@@ -142,10 +140,8 @@ func CopyClose(dst *ContextWriter, src *ContextReader) (err error) {
 				size = int(l.N)
 			}
 		}
-		buf = make([]byte, 0, size)
 	} else {
-		buf = copyBufPool.Get().([]byte)
+		size = cap(buf)
-		defer copyBufPool.Put(buf[:0])
 	}
 	// close both as soon as one of them is done
 	wCloser, wCanClose := dst.Writer.(io.Closer)
@@ -179,7 +175,7 @@ func CopyClose(dst *ContextWriter, src *ContextReader) (err error) {
 	flusher := getHttpFlusher(dst.Writer)
 	canFlush := flusher != nil
 	for {
-		nr, er := src.Reader.Read(buf[:copyBufSize])
+		nr, er := src.Reader.Read(buf[:size])
 		if nr > 0 {
 			nw, ew := dst.Writer.Write(buf[0:nr])
 			if nw < 0 || nr < nw {

internal/utils/pool/pool.go

@@ -3,25 +3,23 @@ package pool
 import (
 	"sort"
 
+	"github.com/puzpuzpuz/xsync/v3"
 	"github.com/yusing/go-proxy/internal/logging"
-	"github.com/yusing/go-proxy/internal/utils"
-	"github.com/yusing/go-proxy/internal/utils/functional"
 )
 
 type (
 	Pool[T Object] struct {
-		m    functional.Map[string, T]
+		m    *xsync.MapOf[string, T]
 		name string
 	}
 	Object interface {
 		Key() string
 		Name() string
-		utils.MapMarshaler
 	}
 )
 
 func New[T Object](name string) Pool[T] {
-	return Pool[T]{functional.NewMapOf[string, T](), name}
+	return Pool[T]{xsync.NewMapOf[string, T](), name}
 }
 
 func (p Pool[T]) Name() string {
@@ -29,6 +27,7 @@ func (p Pool[T]) Name() string {
 }
 
 func (p Pool[T]) Add(obj T) {
+	p.checkExists(obj.Key())
 	p.m.Store(obj.Key(), obj)
 	logging.Info().Msgf("%s: added %s", p.name, obj.Name())
 }
@@ -50,8 +49,8 @@ func (p Pool[T]) Clear() {
 	p.m.Clear()
 }
 
-func (p Pool[T]) Base() functional.Map[string, T] {
+func (p Pool[T]) Iter(fn func(k string, v T) bool) {
-	return p.m
+	p.m.Range(fn)
 }
 
 func (p Pool[T]) Slice() []T {
@@ -64,11 +63,3 @@ func (p Pool[T]) Slice() []T {
 	})
 	return slice
 }
-
-func (p Pool[T]) Iter(fn func(k string, v T) bool) {
-	p.m.Range(fn)
-}
-
-func (p Pool[T]) IterAll(fn func(k string, v T)) {
-	p.m.RangeAll(fn)
-}

internal/utils/pool/pool_debug.go

@@ -0,0 +1,15 @@
+//go:build debug
+
+package pool
+
+import (
+	"runtime/debug"
+
+	"github.com/yusing/go-proxy/internal/logging"
+)
+
+func (p Pool[T]) checkExists(key string) {
+	if _, ok := p.m.Load(key); ok {
+		logging.Warn().Msgf("%s: key %s already exists\nstacktrace: %s", p.name, key, string(debug.Stack()))
+	}
+}

internal/utils/pool/pool_prod.go

@@ -0,0 +1,7 @@
+//go:build !debug
+
+package pool
+
+func (p Pool[T]) checkExists(key string) {
+	// no-op in production
+}

internal/utils/synk/pool.go

@@ -0,0 +1,42 @@
+package synk
+
+import "sync"
+
+type (
+	// Pool is a wrapper of sync.Pool that limits the size of the object.
+	Pool[T any] struct {
+		pool    sync.Pool
+		maxSize int
+	}
+	BytesPool = Pool[byte]
+)
+
+const (
+	DefaultInitBytes = 1024
+	DefaultMaxBytes  = 1024 * 1024
+)
+
+func NewPool[T any](initSize int, maxSize int) *Pool[T] {
+	return &Pool[T]{
+		pool: sync.Pool{
+			New: func() any {
+				return make([]T, 0, initSize)
+			},
+		},
+		maxSize: maxSize,
+	}
+}
+
+func NewBytesPool(initSize int, maxSize int) *BytesPool {
+	return NewPool[byte](initSize, maxSize)
+}
+
+func (p *Pool[T]) Get() []T {
+	return p.pool.Get().([]T)
+}
+
+func (p *Pool[T]) Put(b []T) {
+	if cap(b) <= p.maxSize {
+		p.pool.Put(b[:0])
+	}
+}

internal/watcher/directory_watcher.go

@@ -21,7 +21,7 @@ type DirWatcher struct {
 	w   *fsnotify.Watcher
 
 	fwMap map[string]*fileWatcher
-	mu    sync.Mutex
+	mu    sync.RWMutex
 
 	eventCh chan Event
 	errCh   chan gperr.Error
@@ -56,7 +56,7 @@ func NewDirectoryWatcher(parent task.Parent, dirPath string) *DirWatcher {
 		fwMap:   make(map[string]*fileWatcher),
 		eventCh: make(chan Event),
 		errCh:   make(chan gperr.Error),
-		task:    parent.Subtask("dir_watcher(" + dirPath + ")"),
+		task:    parent.Subtask("dir_watcher("+dirPath+")", true),
 	}
 	go helper.start()
 	return helper
@@ -95,7 +95,9 @@ func (h *DirWatcher) cleanup() {
 		close(fw.eventCh)
 		close(fw.errCh)
 	}
+	h.fwMap = nil
 	h.task.Finish(nil)
+	h.Info().Msg("directory watcher closed")
 }
 
 func (h *DirWatcher) start() {
@@ -143,9 +145,9 @@ func (h *DirWatcher) start() {
 			}
 
 			// send event to file watcher too
-			h.mu.Lock()
+			h.mu.RLock()
 			w, ok := h.fwMap[relPath]
-			h.mu.Unlock()
+			h.mu.RUnlock()
 			if ok {
 				select {
 				case w.eventCh <- msg:

internal/watcher/health/monitor/monitor.go

@@ -11,7 +11,7 @@ import (
 	"github.com/yusing/go-proxy/internal/gperr"
 	"github.com/yusing/go-proxy/internal/logging"
 	"github.com/yusing/go-proxy/internal/notif"
-	route "github.com/yusing/go-proxy/internal/route/types"
+	"github.com/yusing/go-proxy/internal/route/routes"
 	"github.com/yusing/go-proxy/internal/task"
 	"github.com/yusing/go-proxy/internal/utils/atomic"
 	"github.com/yusing/go-proxy/internal/utils/strutils"
@@ -37,15 +37,15 @@ type (
 
 var ErrNegativeInterval = errors.New("negative interval")
 
-func NewMonitor(r route.Route) health.HealthMonCheck {
+func NewMonitor(r routes.Route) health.HealthMonCheck {
 	var mon health.HealthMonCheck
 	if r.IsAgent() {
 		mon = NewAgentProxiedMonitor(r.Agent(), r.HealthCheckConfig(), AgentTargetFromURL(r.TargetURL()))
 	} else {
 		switch r := r.(type) {
-		case route.HTTPRoute:
+		case routes.HTTPRoute:
 			mon = NewHTTPHealthMonitor(r.TargetURL(), r.HealthCheckConfig())
-		case route.StreamRoute:
+		case routes.StreamRoute:
 			mon = NewRawHealthMonitor(r.TargetURL(), r.HealthCheckConfig())
 		default:
 			logging.Panic().Msgf("unexpected route type: %T", r)
@@ -58,7 +58,7 @@ func NewMonitor(r route.Route) health.HealthMonCheck {
 			return mon
 		}
 		r.Task().OnCancel("close_docker_client", client.Close)
-		return NewDockerHealthMonitor(client, cont.ContainerID, r.TargetName(), r.HealthCheckConfig(), mon)
+		return NewDockerHealthMonitor(client, cont.ContainerID, r.Name(), r.HealthCheckConfig(), mon)
 	}
 	return mon
 }
@@ -88,7 +88,7 @@ func (mon *monitor) Start(parent task.Parent) gperr.Error {
 	}
 
 	mon.service = parent.Name()
-	mon.task = parent.Subtask("health_monitor")
+	mon.task = parent.Subtask("health_monitor", true)
 
 	go func() {
 		logger := logging.With().Str("name", mon.service).Logger()

pkg/json/json.go

@@ -2,9 +2,9 @@ package json
 
 import (
 	"reflect"
-	"sync"
 
 	"github.com/bytedance/sonic"
+	"github.com/yusing/go-proxy/internal/utils/synk"
 )
 
 type Marshaler interface {
@@ -38,8 +38,8 @@ var (
 //
 //   - It does not support maps other than string-keyed maps.
 func Marshal(v any) ([]byte, error) {
-	buf := newBytes()
+	buf := bytesPool.Get()
-	defer putBytes(buf)
+	defer bytesPool.Put(buf)
 	return cloneBytes(appendMarshal(reflect.ValueOf(v), buf)), nil
 }
 
@@ -47,21 +47,9 @@ func MarshalTo(v any, buf []byte) []byte {
 	return appendMarshal(reflect.ValueOf(v), buf)
 }
 
-const bufSize = 8192
+const initBufSize = 4096
 
-var bytesPool = sync.Pool{
+var bytesPool = synk.NewBytesPool(initBufSize, synk.DefaultMaxBytes)
-	New: func() any {
-		return make([]byte, 0, bufSize)
-	},
-}
-
-func newBytes() []byte {
-	return bytesPool.Get().([]byte)
-}
-
-func putBytes(buf []byte) {
-	bytesPool.Put(buf[:0])
-}
 
 func cloneBytes(buf []byte) (res []byte) {
 	return append(res, buf...)