fix(triple): move autoScaler to be triggered by Get/Put and fix naming

Author: Aetherance

protocol/triple/client.go

@@ -312,7 +312,7 @@ func newClientManager(url *common.URL) (*clientManager, error) {
 		baseTriURL = httpPrefix + baseTriURL
 	}
 
-	triPools := make(map[string]*TriClientPool)
+	triClientPools := make(map[string]*TriClientPool)
 
 	if len(url.Methods) != 0 {
 		for _, method := range url.Methods {
@@ -323,7 +323,7 @@ func newClientManager(url *common.URL) (*clientManager, error) {
 			pool := NewTriClientPool(triClientPoolMaxSize, func() *tri.Client {
 				return tri.NewClient(httpClient, triURL, cliOpts...)
 			})
-			triPools[method] = pool
+			triClientPools[method] = pool
 		}
 	} else {
 		// This branch is for the non-IDL mode, where we pass in the service solely
@@ -336,21 +336,21 @@ func newClientManager(url *common.URL) (*clientManager, error) {
 
 		serviceType := reflect.TypeOf(service)
 		for i := range serviceType.NumMethod() {
-			method := serviceType.Method(i).Name
-			triURL, err := joinPath(baseTriURL, url.Interface(), method)
+			methodName := serviceType.Method(i).Name
+			triURL, err := joinPath(baseTriURL, url.Interface(), methodName)
 			if err != nil {
-				return nil, fmt.Errorf("failed to join path for method %s: %w", method, err)
+				return nil, fmt.Errorf("failed to join path for method %s: %w", methodName, err)
 			}
 			pool := NewTriClientPool(triClientPoolMaxSize, func() *tri.Client {
 				return tri.NewClient(httpClient, triURL, cliOpts...)
 			})
-			triPools[method] = pool
+			triClientPools[methodName] = pool
 		}
 	}
 
 	return &clientManager{
 		isIDL:      isIDL,
-		triClients: triPools,
+		triClients: triClientPools,
 	}, nil
 }
 

protocol/triple/client_pool.go

@@ -43,10 +43,9 @@ var (
 )
 
 const (
-	autoScalerPeriod        = 700 * time.Millisecond // autoScaler interval
-	maxExpandPerCycle       = 16                     // maximum number of clients to expand per cycle
-	lowIdleThresholdDivisor = 5                      // consider clients insufficient if idle < curSize/5
-	highIdleStreakLimit     = 10                     // consecutive high idle count to trigger shrinking
+	maxExpandPerCycle       = 16 // maximum number of clients to expand per cycle
+	lowIdleThresholdDivisor = 5  // consider clients insufficient if idle < curSize/5
+	highIdleStreakLimit     = 10 // consecutive high idle count to trigger shrinking
 )
 
 type TriClientPool struct {
@@ -58,7 +57,8 @@ type TriClientPool struct {
 	closed      bool
 	getTimeouts int // recent timeout count, used to trigger expansion
 
-	fallback *tri.Client
+	fallback            *tri.Client
+	consecutiveHighIdle int
 }
 
 func NewTriClientPool(maxSize int, factory func() *tri.Client) *TriClientPool {
@@ -67,7 +67,6 @@ func NewTriClientPool(maxSize int, factory func() *tri.Client) *TriClientPool {
 		factory: factory,
 		maxSize: maxSize,
 	}
-	go pool.autoScaler()
 	return pool
 }
 
@@ -80,6 +79,8 @@ func NewTriClientPool(maxSize int, factory func() *tri.Client) *TriClientPool {
 // If expansion is not allowed, it waits for a client up to timeout.
 // After timeout, if still no client, Get() returns ErrTimeout with fallback.
 func (p *TriClientPool) Get(timeout time.Duration) (*tri.Client, error) {
+	p.autoScaler()
+
 	select {
 	case cli := <-p.clients:
 		return cli, nil
@@ -122,6 +123,8 @@ func (p *TriClientPool) Get(timeout time.Duration) (*tri.Client, error) {
 // If it fails, Put will drop the client and notify the pool.
 // Dropping a client is part of shrinking.
 func (p *TriClientPool) Put(c *tri.Client) {
+	p.autoScaler()
+
 	if c == nil {
 		return
 	}
@@ -168,38 +171,33 @@ func (p *TriClientPool) Closed() bool {
 	return p.closed
 }
 
-// autoScaler is the main worker that adjusts pool size.
-// It checks regularly whether pool should expand or shrink.
+// autoScaler performs scaling check.
+// It is triggered by Get() or Put()
 // If the timeout count is high, autoScaler tends to expand.
 // If the idle client count is often high, autoScaler tends to shrink.
 func (p *TriClientPool) autoScaler() {
-	ticker := time.NewTicker(autoScalerPeriod)
-	defer ticker.Stop()
-	consecutiveHighIdle := 0
+	p.mu.Lock()
+	defer p.mu.Unlock()
 
-	for range ticker.C {
-		if p.Closed() {
-			return
-		}
+	if p.closed {
+		return
+	}
 
-		p.mu.Lock()
-		curSize := p.curSize
-		idle := len(p.clients)
-		timeouts := p.getTimeouts
-		p.getTimeouts = 0
-		p.mu.Unlock()
+	curSize := p.curSize
+	idle := len(p.clients)
+	timeouts := p.getTimeouts
+	p.getTimeouts = 0
 
-		needExpand := checkExpand(curSize, idle, timeouts)
-		if needExpand != 0 {
-			p.expand(needExpand)
-			consecutiveHighIdle = 0
-			continue
-		}
+	needExpand := checkExpand(curSize, idle, timeouts)
+	if needExpand != 0 {
+		p.expand(needExpand)
+		p.consecutiveHighIdle = 0
+		return
+	}
 
-		needShrink := checkShrink(curSize, idle, &consecutiveHighIdle)
-		if needShrink != 0 {
-			p.shrink(needShrink)
-		}
+	needShrink := checkShrink(curSize, idle, &p.consecutiveHighIdle)
+	if needShrink != 0 {
+		p.shrink(needShrink)
 	}
 }