refactor(adaptive_throttler, resource_monitor): implement shared monitor registry with reference counting

- Replace singleton ResourceMonitor with shared registry pattern for efficient resource sharing
- Add dynamic interval updates and improved CPU mode switching
- Enhance memory sampling with better fallback handling
- Add InDelta assertion helper for precise float comparisons
- Comprehensive test refactoring with improved mocking and coverage
- Update adaptive throttler tests to use new resource monitor architecture

Author: kxrxh

Dockerfile.test

@@ -0,0 +1,17 @@
+FROM golang:1.21
+
+WORKDIR /app
+
+# Copy go mod files (go.sum may not exist)
+COPY go.mod ./
+COPY */go.mod ./
+
+# Download dependencies
+RUN go mod download
+
+# Copy source code
+COPY . .
+
+# Run tests with coverage
+CMD ["sh", "-c", "echo '=== Running all sysmonitor tests with coverage ===' && go test -v -coverprofile=sysmonitor_coverage.out -covermode=atomic ./internal/sysmonitor && echo '' && echo '=== sysmonitor coverage ===' && go tool cover -func=sysmonitor_coverage.out && echo '' && echo '=== Running adaptive_throttler and resource_monitor tests with coverage ===' && go test -v -coverprofile=flow_coverage.out -covermode=atomic ./flow -run 'TestAdaptiveThrottler|TestResourceMonitor|TestInitSampler|TestClampPercent|TestValidatePercent|TestValidateResourceStats|TestMemoryUsagePercent' && echo '' && echo '=== flow coverage (adaptive_throttler + resource_monitor) ===' && go tool cover -func=flow_coverage.out | grep -E '(adaptive_throttler|resource_monitor|initSampler|clampPercent|validatePercent|validateResourceStats|memoryUsagePercent|total)' && echo '' && echo '=== Combined coverage summary ===' && go tool cover -func=sysmonitor_coverage.out | tail -1 && go tool cover -func=flow_coverage.out | tail -1"]
+

examples/adaptive_throttler/demo/demo.go

@@ -14,20 +14,18 @@ import (
 // The demo:
 // 1. Produces 250 elements in bursts
 // 2. Processes elements with CPU-intensive work (50ms each)
-// 3. Simulates increasing memory pressure as more elements are processed
+// 3. Simulates memory pressure that increases then decreases (creating throttling-recovery cycle)
 // 4. The adaptive throttler adjusts throughput based on CPU/memory usage
-// 5. Stats are logged every 500ms showing rate adaptation
+// 5. Shows throttling down to ~1/sec during high memory, then recovery back to 40/sec
+// 6. Stats are logged every 500ms showing rate adaptation
 func main() {
 	var elementsProcessed atomic.Int64
 
 	// Set up demo configuration with memory simulation
 	throttler := setupDemoThrottler(&elementsProcessed)
-	defer func() {
-		throttler.Close()
-	}()
 
 	in := make(chan any)
-	out := make(chan any, 32)
+	out := make(chan any) // Unbuffered channel to prevent apparent bursts
 
 	source := ext.NewChanSource(in)
 	sink := ext.NewChanSink(out)
@@ -45,62 +43,64 @@ func main() {
 
 	go produceBurst(in, 250)
 
-	// Use a variable to prevent compiler optimization of CPU work
 	var cpuWorkChecksum uint64
 
+	// Process the output
+	elementsReceived := 0
 	for element := range sink.Out {
 		fmt.Printf("consumer received %v\n", element)
-		elementsProcessed.Add(1) // Track processed elements for memory pressure simulation
+		elementsProcessed.Add(1)
+		elementsReceived++
 
-		// Perform CPU-intensive work that can't be optimized away
-		// This ensures Windows GetProcessTimes can detect CPU usage
-		// (Windows timer resolution is ~15.625ms, so we need at least 50-100ms of work)
+		// Perform CPU-intensive work
 		burnCPU(50*time.Millisecond, &cpuWorkChecksum)
 
 		time.Sleep(25 * time.Millisecond)
 	}
 
-	// Print checksum to ensure CPU work wasn't optimized away
 	fmt.Printf("CPU work checksum: %d\n", cpuWorkChecksum)
-
+	fmt.Printf("Total elements produced: 250, Total elements received: %d\n", elementsReceived)
+	if elementsReceived == 250 {
+		fmt.Println("✅ SUCCESS: All elements processed without dropping!")
+	} else {
+		fmt.Printf("❌ FAILURE: %d elements were dropped!\n", 250-elementsReceived)
+	}
 	fmt.Println("adaptive throttling pipeline completed")
 }
 
 // setupDemoThrottler creates and configures an adaptive throttler with demo settings
 func setupDemoThrottler(elementsProcessed *atomic.Int64) *flow.AdaptiveThrottler {
 	config := flow.DefaultAdaptiveThrottlerConfig()
-	config.MinThroughput = 5
-	config.MaxThroughput = 40
+
+	config.MinRate = 1
+	config.MaxRate = 20
+	config.InitialRate = 20
 	config.SampleInterval = 200 * time.Millisecond
-	config.BufferSize = 32
-	config.AdaptationFactor = 0.5
-	config.SmoothTransitions = true
-	config.MaxMemoryPercent = 40.0
-	config.MaxCPUPercent = 80.0
 
+	config.BackoffFactor = 0.5
+	config.RecoveryFactor = 1.5
+
+	config.MaxMemoryPercent = 35.0
+	config.RecoveryMemoryThreshold = 30.0
+
+	// Memory Reader Simulation - Creates a cycle: low -> high -> low memory usage
 	config.MemoryReader = func() (float64, error) {
 		elementCount := elementsProcessed.Load()
 
-		// Memory pressure increases with processed elements:
-		// - 0-50 elements: 5% memory
-		// - 51-100 elements: 15% memory
-		// - 101-150 elements: 30% memory
-		// - 151+ elements: 50%+ memory (increases gradually)
 		var memoryPercent float64
 		switch {
-		case elementCount <= 50:
-			memoryPercent = 5.0 + float64(elementCount)*0.2 // 5% to 15%
-		case elementCount <= 100:
-			memoryPercent = 15.0 + float64(elementCount-50)*0.3 // 15% to 30%
-		case elementCount <= 150:
-			memoryPercent = 30.0 + float64(elementCount-100)*0.4 // 30% to 50%
-		default:
-			memoryPercent = 50.0 + float64(elementCount-150)*0.3 // 50%+ (increases more slowly)
-			if memoryPercent > 95.0 {
-				memoryPercent = 95.0
+		case elementCount <= 80: // Phase 1: Low memory, allow high throughput
+			memoryPercent = 5.0 + float64(elementCount)*0.1 // 5% to 13%
+		case elementCount <= 120: // Phase 2: Increasing memory pressure, cause throttling
+			memoryPercent = 15.0 + float64(elementCount-80)*0.6 // 15% to 43%
+		case elementCount <= 160: // Phase 3: High memory, keep throttled down to ~1/sec
+			memoryPercent = 30.0 + float64(elementCount-120)*0.3 // 30% to 42%
+		default: // Phase 4: Memory decreases, allow recovery back to 40/sec
+			memoryPercent = 25.0 - float64(elementCount-160)*1.5 // 25% down to ~5%
+			if memoryPercent < 5.0 {
+				memoryPercent = 5.0
 			}
 		}
-
 		return memoryPercent, nil
 	}
 
@@ -114,20 +114,17 @@ func setupDemoThrottler(elementsProcessed *atomic.Int64) *flow.AdaptiveThrottler
 func produceBurst(in chan<- any, total int) {
 	defer close(in)
 
-	for i := 0; i < total; i++ {
+	for i := range total {
 		in <- fmt.Sprintf("job-%02d", i)
 
 		if (i+1)%10 == 0 {
 			time.Sleep(180 * time.Millisecond)
 			continue
 		}
-
 		time.Sleep(time.Duration(2+rand.Intn(5)) * time.Millisecond)
 	}
 }
 
-// burnCPU performs CPU-intensive work for the specified duration
-// The checksum parameter prevents the compiler from optimizing away the work
 func burnCPU(duration time.Duration, checksum *uint64) {
 	start := time.Now()
 	for time.Since(start) < duration {
@@ -147,8 +144,8 @@ func logThrottlerStats(at *flow.AdaptiveThrottler, done <-chan struct{}) {
 			return
 		case <-ticker.C:
 			stats := at.GetResourceStats()
-			fmt.Printf("[stats] rate=%d eps memory=%.1f%% cpu=%.1f%% goroutines=%d\n",
-				at.GetCurrentRate(), stats.MemoryUsedPercent, stats.CPUUsagePercent, stats.GoroutineCount)
+			fmt.Printf("[stats] Rate: %.1f/sec, CPU: %.1f%%, Memory: %.1f%%\n",
+				at.GetCurrentRate(), stats.CPUUsagePercent, stats.MemoryUsedPercent)
 		}
 	}
 }

examples/adaptive_throttler/main.go

@@ -13,33 +13,78 @@ func editMessage(msg string) string {
 	return strings.ToUpper(msg)
 }
 
+func addTimestamp(msg string) string {
+	return fmt.Sprintf("[%s] %s", time.Now().Format("15:04:05"), msg)
+}
+
 func main() {
-	// To setup a custom throttler, you can modify the throttlerConfig struct with your desired values.
-	// For all available options, see the flow.AdaptiveThrottlerConfig struct.
-	throttlerConfig := flow.DefaultAdaptiveThrottlerConfig()
-	throttler, err := flow.NewAdaptiveThrottler(throttlerConfig)
+	// Configure first throttler - focuses on CPU limits
+	throttler1Config := flow.DefaultAdaptiveThrottlerConfig()
+	throttler1Config.MaxCPUPercent = 70.0
+	throttler1Config.MaxMemoryPercent = 60.0
+	throttler1Config.InitialRate = 15
+	throttler1Config.MaxRate = 30
+	throttler1Config.SampleInterval = 200 * time.Millisecond
+	throttler1Config.CPUUsageMode = flow.CPUUsageModeMeasured // Use heuristic for better CPU visibility
+
+	throttler1, err := flow.NewAdaptiveThrottler(throttler1Config)
+	if err != nil {
+		panic(fmt.Sprintf("failed to create throttler1: %v", err))
+	}
+
+	// Configure second throttler - focuses on memory limits
+	throttler2Config := flow.DefaultAdaptiveThrottlerConfig()
+	throttler2Config.MaxCPUPercent = 75.0
+	throttler2Config.MaxMemoryPercent = 50.0
+	throttler2Config.InitialRate = 10
+	throttler2Config.MaxRate = 25
+	throttler2Config.SampleInterval = 200 * time.Millisecond
+	throttler2Config.CPUUsageMode = flow.CPUUsageModeMeasured // Use heuristic for better CPU visibility
+
+	throttler2, err := flow.NewAdaptiveThrottler(throttler2Config)
 	if err != nil {
-		panic(fmt.Sprintf("failed to create adaptive throttler: %v", err))
+		panic(fmt.Sprintf("failed to create throttler2: %v", err))
 	}
-	defer throttler.Close()
 
 	in := make(chan any)
 
 	source := ext.NewChanSource(in)
 	editMapFlow := flow.NewMap(editMessage, 1)
+	timestampFlow := flow.NewMap(addTimestamp, 1)
 	sink := ext.NewStdoutSink()
 
+	// Pipeline: Source -> Throttler1 -> Edit -> Throttler2 -> Timestamp -> Sink
+	go func() {
+		source.
+			Via(throttler1).
+			Via(editMapFlow).
+			Via(throttler2).
+			Via(timestampFlow).
+			To(sink)
+	}()
+
+	// Stats logging goroutine
 	go func() {
-		source.Via(throttler).Via(editMapFlow).To(sink)
+		ticker := time.NewTicker(1 * time.Second)
+		defer ticker.Stop()
+
+		for range ticker.C {
+			stats := throttler1.GetResourceStats()
+			fmt.Printf("[stats] T1-Rate: %.1f/s, T2-Rate: %.1f/s, CPU: %.1f%%, Mem: %.1f%%\n",
+				throttler1.GetCurrentRate(),
+				throttler2.GetCurrentRate(),
+				stats.CPUUsagePercent,
+				stats.MemoryUsedPercent)
+		}
 	}()
 
 	go func() {
 		defer close(in)
 
-		for i := 1; i <= 50; i++ {
+		for i := 1; i <= 100; i++ {
 			message := fmt.Sprintf("message-%d", i)
 			in <- message
-			time.Sleep(50 * time.Millisecond)
+			time.Sleep(30 * time.Millisecond)
 		}
 	}()