Merge pull request #180 from deploymenttheory/dev

Adjust concurrency logic and metrics in the concurrency package, and …

Author: ShocOne

concurrency/handler.go

@@ -35,12 +35,13 @@ type ConcurrencyMetrics struct {
 		Lock  sync.Mutex // Lock for throughput metrics/
 	}
 	ResponseTimeVariability struct { // Metrics related to response time variability
-		Total           time.Duration // Total response time for all requests
-		Average         time.Duration // Average response time across all requests
-		Variance        float64       // Variance of response times
-		Count           int64         // Count of responses used for calculating response time variability
-		Lock            sync.Mutex    // Lock for response time variability metrics
-		StdDevThreshold float64       // Maximum acceptable standard deviation for adjusting concurrency
+		Total                  time.Duration // Total response time for all requests
+		Average                time.Duration // Average response time across all requests
+		Variance               float64       // Variance of response times
+		Count                  int64         // Count of responses used for calculating response time variability
+		Lock                   sync.Mutex    // Lock for response time variability metrics
+		StdDevThreshold        float64       // Maximum acceptable standard deviation for adjusting concurrency
+		DebounceScaleDownCount int           // Counter to manage scale down actions after consecutive triggers
 	}
 	ResponseCodeMetrics struct {
 		ErrorRate float64    // Error rate calculated as (TotalRateLimitErrors + 5xxErrors) / TotalRequests

concurrency/metrics.go

@@ -199,11 +199,11 @@ var responseTimes []time.Duration
 func (ch *ConcurrencyHandler) MonitorResponseTimeVariability(responseTime time.Duration) int {
 	// Append the latest response time
 	responseTimes = append(responseTimes, responseTime)
-	if len(responseTimes) > 5 { // Use the last 5 measurements
+	if len(responseTimes) > 10 { // Use the last 10 measurements for a smoother average
 		responseTimes = responseTimes[1:]
 	}
 
-	// Calculate average response time from the slice
+	// Calculate moving average response time from the slice
 	var sum time.Duration
 	for _, rt := range responseTimes {
 		sum += rt
@@ -218,11 +218,20 @@ func (ch *ConcurrencyHandler) MonitorResponseTimeVariability(responseTime time.D
 	variance := varianceSum / float64(len(responseTimes))
 	stdDev := math.Sqrt(variance)
 
-	// Determine action based on standard deviation against a higher threshold
-	if stdDev > (ch.Metrics.ResponseTimeVariability.StdDevThreshold * 1.5) { // Increased threshold
-		return -1 // Suggest decrease concurrency
-	} else if stdDev <= ch.Metrics.ResponseTimeVariability.StdDevThreshold && len(ch.sem) < MaxConcurrency {
-		return 1 // Suggest increase concurrency if there is capacity
+	// Action determination with debounce effect
+	// Requires keeping track of how often the stdDev threshold has been exceeded
+	const debounceCount = 3 // e.g., threshold must be exceeded in 3 consecutive checks to act
+	if stdDev > (ch.Metrics.ResponseTimeVariability.StdDevThreshold * 1.5) {
+		ch.Metrics.ResponseTimeVariability.DebounceScaleDownCount++
+		if ch.Metrics.ResponseTimeVariability.DebounceScaleDownCount >= debounceCount {
+			ch.Metrics.ResponseTimeVariability.DebounceScaleDownCount = 0 // reset counter after action
+			return -1                                                     // Suggest decrease concurrency
+		}
+	} else {
+		ch.Metrics.ResponseTimeVariability.DebounceScaleDownCount = 0 // reset counter if condition not met
+		if stdDev <= ch.Metrics.ResponseTimeVariability.StdDevThreshold && len(ch.sem) < MaxConcurrency {
+			return 1 // Suggest increase concurrency if there is capacity
+		}
 	}
 	return 0
 }