Enhance caching mechanisms and introduce adaptive features

- Updated caching logic to support size-based promotion filtering, ensuring that not all files may be promoted based on size constraints. - Implemented adaptive caching strategies with a new AdaptiveCacheManager to analyze access patterns and adjust caching strategies dynamically. - Introduced predictive caching features with a PredictiveCacheManager to prefetch content based on access patterns. - Added a CacheWarmer to preload popular content into the cache, improving access times for frequently requested files. - Refactored memory management with a DynamicCacheManager to adjust cache sizes based on system memory usage. - Enhanced VFS interface and file metadata handling to support new features and improve performance. - Updated tests to validate new caching behaviors and ensure reliability of the caching system.
2025-09-21 22:47:13 -05:00
parent bbe014e334
commit 45ae234694
12 changed files with 2212 additions and 189 deletions
--- a/steamcache/steamcache.go
+++ b/steamcache/steamcache.go
@@ -16,10 +16,13 @@ import (
 	"regexp"
 	"s1d3sw1ped/SteamCache2/steamcache/logger"
 	"s1d3sw1ped/SteamCache2/vfs"
+	"s1d3sw1ped/SteamCache2/vfs/adaptive"
 	"s1d3sw1ped/SteamCache2/vfs/cache"
 	"s1d3sw1ped/SteamCache2/vfs/disk"
 	"s1d3sw1ped/SteamCache2/vfs/gc"
 	"s1d3sw1ped/SteamCache2/vfs/memory"
+	"s1d3sw1ped/SteamCache2/vfs/predictive"
+	"s1d3sw1ped/SteamCache2/vfs/warming"
 	"strconv"
 	"strings"
 	"sync"
@@ -771,14 +774,20 @@ type coalescedRequest struct {
 	waitingCount int
 	done         bool
 	mu           sync.Mutex
+	// Buffered response data for coalesced clients
+	responseData    []byte
+	responseHeaders http.Header
+	statusCode      int
+	status          string
 }

 func newCoalescedRequest() *coalescedRequest {
 	return &coalescedRequest{
-		responseChan: make(chan *http.Response, 1),
-		errorChan:    make(chan error, 1),
-		waitingCount: 1,
-		done:         false,
+		responseChan:    make(chan *http.Response, 1),
+		errorChan:       make(chan error, 1),
+		waitingCount:    1,
+		done:            false,
+		responseHeaders: make(http.Header),
 	}
 }

@@ -802,6 +811,17 @@ func (cr *coalescedRequest) complete(resp *http.Response, err error) {
 		default:
 		}
 	} else {
+		// Store response data for coalesced clients
+		if resp != nil {
+			cr.statusCode = resp.StatusCode
+			cr.status = resp.Status
+			// Copy headers (excluding hop-by-hop headers)
+			for k, vv := range resp.Header {
+				if _, skip := hopByHopHeaders[http.CanonicalHeaderKey(k)]; !skip {
+					cr.responseHeaders[k] = vv
+				}
+			}
+		}
 		select {
 		case cr.responseChan <- resp:
 		default:
@@ -809,6 +829,14 @@ func (cr *coalescedRequest) complete(resp *http.Response, err error) {
 	}
 }

+// setResponseData stores the buffered response data for coalesced clients
+func (cr *coalescedRequest) setResponseData(data []byte) {
+	cr.mu.Lock()
+	defer cr.mu.Unlock()
+	cr.responseData = make([]byte, len(data))
+	copy(cr.responseData, data)
+}
+
 // getOrCreateCoalescedRequest gets an existing coalesced request or creates a new one
 func (sc *SteamCache) getOrCreateCoalescedRequest(cacheKey string) (*coalescedRequest, bool) {
 	sc.coalescedRequestsMu.Lock()
@@ -899,8 +927,8 @@ type SteamCache struct {
 	memory *memory.MemoryFS
 	disk   *disk.DiskFS

-	memorygc *gc.GCFS
-	diskgc   *gc.GCFS
+	memorygc *gc.AsyncGCFS
+	diskgc   *gc.AsyncGCFS

 	server *http.Server
 	client *http.Client
@@ -922,6 +950,18 @@ type SteamCache struct {

 	// Service management
 	serviceManager *ServiceManager
+
+	// Adaptive and predictive caching
+	adaptiveManager   *adaptive.AdaptiveCacheManager
+	predictiveManager *predictive.PredictiveCacheManager
+	cacheWarmer       *warming.CacheWarmer
+	lastAccessKey     string // Track last accessed key for sequence analysis
+	lastAccessKeyMu   sync.RWMutex
+	adaptiveEnabled   bool // Flag to enable/disable adaptive features
+
+	// Dynamic memory management
+	memoryMonitor   *memory.MemoryMonitor
+	dynamicCacheMgr *memory.DynamicCacheManager
 }

 func New(address string, memorySize string, diskSize string, diskPath, upstream, memoryGC, diskGC string, maxConcurrentRequests int64, maxRequestsPerClient int64) *SteamCache {
@@ -938,25 +978,27 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
 	c := cache.New()

 	var m *memory.MemoryFS
-	var mgc *gc.GCFS
+	var mgc *gc.AsyncGCFS
 	if memorysize > 0 {
 		m = memory.New(memorysize)
 		memoryGCAlgo := gc.GCAlgorithm(memoryGC)
 		if memoryGCAlgo == "" {
 			memoryGCAlgo = gc.LRU // default to LRU
 		}
-		mgc = gc.New(m, memoryGCAlgo)
+		// Use hybrid async GC with thresholds: 80% async, 95% sync, 100% hard limit
+		mgc = gc.NewAsync(m, memoryGCAlgo, true, 0.8, 0.95, 1.0)
 	}

 	var d *disk.DiskFS
-	var dgc *gc.GCFS
+	var dgc *gc.AsyncGCFS
 	if disksize > 0 {
 		d = disk.New(diskPath, disksize)
 		diskGCAlgo := gc.GCAlgorithm(diskGC)
 		if diskGCAlgo == "" {
 			diskGCAlgo = gc.LRU // default to LRU
 		}
-		dgc = gc.New(d, diskGCAlgo)
+		// Use hybrid async GC with thresholds: 80% async, 95% sync, 100% hard limit
+		dgc = gc.NewAsync(d, diskGCAlgo, true, 0.8, 0.95, 1.0)
 	}

 	// configure the cache to match the specified mode (memory only, disk only, or memory and disk) based on the provided sizes
@@ -980,23 +1022,48 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
 	}

 	transport := &http.Transport{
-		MaxIdleConns:        200,               // Increased from 100
-		MaxIdleConnsPerHost: 50,                // Increased from 10
-		IdleConnTimeout:     120 * time.Second, // Increased from 90s
+		// Connection pooling optimizations
+		MaxIdleConns:        500,               // Increased for high concurrency
+		MaxIdleConnsPerHost: 100,               // Increased for better connection reuse
+		MaxConnsPerHost:     200,               // Limit connections per host
+		IdleConnTimeout:     300 * time.Second, // Longer idle timeout for better reuse
+
+		// Dial optimizations
 		DialContext: (&net.Dialer{
-			Timeout:   30 * time.Second,
-			KeepAlive: 30 * time.Second,
+			Timeout:   10 * time.Second, // Faster connection timeout
+			KeepAlive: 60 * time.Second, // Longer keep-alive
+			DualStack: true,             // Enable dual-stack (IPv4/IPv6)
 		}).DialContext,
-		TLSHandshakeTimeout:   15 * time.Second, // Increased from 10s
-		ResponseHeaderTimeout: 30 * time.Second, // Increased from 10s
-		ExpectContinueTimeout: 5 * time.Second,  // Increased from 1s
-		DisableCompression:    true,             // Steam doesn't use compression
-		ForceAttemptHTTP2:     true,             // Enable HTTP/2 if available
+
+		// Timeout optimizations
+		TLSHandshakeTimeout:   5 * time.Second,  // Faster TLS handshake
+		ResponseHeaderTimeout: 15 * time.Second, // Faster header timeout
+		ExpectContinueTimeout: 1 * time.Second,  // Faster expect-continue
+
+		// Performance optimizations
+		DisableCompression: true,  // Steam doesn't use compression
+		ForceAttemptHTTP2:  true,  // Enable HTTP/2 if available
+		DisableKeepAlives:  false, // Enable keep-alives
+
+		// Buffer optimizations
+		WriteBufferSize: 64 * 1024, // 64KB write buffer
+		ReadBufferSize:  64 * 1024, // 64KB read buffer
+
+		// Connection reuse optimizations
+		MaxResponseHeaderBytes: 1 << 20, // 1MB max header size
 	}

 	client := &http.Client{
 		Transport: transport,
-		Timeout:   120 * time.Second, // Increased from 60s
+		Timeout:   60 * time.Second, // Optimized timeout for better responsiveness
+		// Add redirect policy for better performance
+		CheckRedirect: func(req *http.Request, via []*http.Request) error {
+			// Limit redirects to prevent infinite loops
+			if len(via) >= 10 {
+				return http.ErrUseLastResponse
+			}
+			return nil
+		},
 	}

 	sc := &SteamCache{
@@ -1010,11 +1077,12 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
 		client:   client,
 		server: &http.Server{
 			Addr:              address,
-			ReadTimeout:       30 * time.Second,  // Increased
-			WriteTimeout:      60 * time.Second,  // Increased
-			IdleTimeout:       120 * time.Second, // Good for keep-alive
-			ReadHeaderTimeout: 10 * time.Second,  // New, for header attacks
-			MaxHeaderBytes:    1 << 20,           // 1MB, optional
+			ReadTimeout:       15 * time.Second,  // Optimized for faster response
+			WriteTimeout:      30 * time.Second,  // Optimized for faster response
+			IdleTimeout:       300 * time.Second, // Longer idle timeout for better connection reuse
+			ReadHeaderTimeout: 5 * time.Second,   // Faster header timeout
+			MaxHeaderBytes:    1 << 20,           // 1MB max header size
+			// Connection optimizations will be handled in ServeHTTP method
 		},

 		// Initialize concurrency control fields
@@ -1026,6 +1094,23 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,

 		// Initialize service management
 		serviceManager: NewServiceManager(),
+
+		// Initialize adaptive and predictive caching (lightweight)
+		adaptiveManager:   adaptive.NewAdaptiveCacheManager(5 * time.Minute), // Much longer interval
+		predictiveManager: predictive.NewPredictiveCacheManager(),
+		cacheWarmer:       warming.NewCacheWarmer(c, 2), // Reduced to 2 concurrent warmers
+		adaptiveEnabled:   true,                         // Enable by default but can be disabled
+
+		// Initialize dynamic memory management
+		memoryMonitor:   memory.NewMemoryMonitor(uint64(memorysize), 10*time.Second, 0.1), // 10% threshold
+		dynamicCacheMgr: nil,                                                              // Will be set after cache creation
+	}
+
+	// Initialize dynamic cache manager if we have memory cache
+	if m != nil && sc.memoryMonitor != nil {
+		sc.dynamicCacheMgr = memory.NewDynamicCacheManager(mgc, uint64(memorysize), sc.memoryMonitor)
+		sc.dynamicCacheMgr.Start()
+		sc.memoryMonitor.Start()
 	}

 	// Log GC algorithm configuration
@@ -1090,6 +1175,10 @@ func (sc *SteamCache) Shutdown() {
 }

 func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	// Set keep-alive headers for better performance
+	w.Header().Set("Connection", "keep-alive")
+	w.Header().Set("Keep-Alive", "timeout=300, max=1000")
+
 	// Apply global concurrency limit first
 	if err := sc.requestSemaphore.Acquire(context.Background(), 1); err != nil {
 		logger.Logger.Warn().Str("client_ip", getClientIP(r)).Msg("Server at capacity, rejecting request")
@@ -1192,7 +1281,9 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 						Msg("Failed to deserialize cache file - removing corrupted entry")
 					sc.vfs.Delete(cachePath)
 				} else {
-					// Cache validation passed
+					// Cache validation passed - record access for adaptive/predictive analysis
+					sc.recordCacheAccess(cacheKey, int64(len(cachedData)))
+
 					logger.Logger.Debug().
 						Str("key", cacheKey).
 						Str("url", urlPath).
@@ -1220,54 +1311,43 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {

 			select {
 			case resp := <-coalescedReq.responseChan:
-				// Use the downloaded response
+				// Use the buffered response data instead of making a fresh request
 				defer resp.Body.Close()

-				// For coalesced clients, we need to make a new request to get fresh data
-				// since the original response body was consumed by the first client
-				freshReq, err := http.NewRequest(http.MethodGet, r.URL.String(), nil)
-				if err != nil {
+				// Wait for response data to be available
+				coalescedReq.mu.Lock()
+				for coalescedReq.responseData == nil && coalescedReq.done {
+					coalescedReq.mu.Unlock()
+					time.Sleep(1 * time.Millisecond) // Brief wait for data to be set
+					coalescedReq.mu.Lock()
+				}
+
+				if coalescedReq.responseData == nil {
+					coalescedReq.mu.Unlock()
 					logger.Logger.Error().
-						Err(err).
 						Str("key", cacheKey).
 						Str("url", urlPath).
 						Str("client_ip", clientIP).
-						Msg("Failed to create fresh request for coalesced client")
-					http.Error(w, "Failed to fetch data", http.StatusInternalServerError)
+						Msg("No response data available for coalesced client")
+					http.Error(w, "No response data available", http.StatusInternalServerError)
 					return
 				}

-				// Copy original headers
-				for k, vv := range r.Header {
-					freshReq.Header[k] = vv
-				}
+				// Copy the buffered response data
+				responseData := make([]byte, len(coalescedReq.responseData))
+				copy(responseData, coalescedReq.responseData)
+				coalescedReq.mu.Unlock()

-				freshResp, err := sc.client.Do(freshReq)
-				if err != nil {
-					logger.Logger.Error().
-						Err(err).
-						Str("key", cacheKey).
-						Str("url", urlPath).
-						Str("client_ip", clientIP).
-						Msg("Failed to fetch fresh data for coalesced client")
-					http.Error(w, "Failed to fetch data", http.StatusInternalServerError)
-					return
-				}
-				defer freshResp.Body.Close()
-
-				// Serve the fresh response
-				for k, vv := range freshResp.Header {
-					if _, skip := hopByHopHeaders[http.CanonicalHeaderKey(k)]; skip {
-						continue
-					}
+				// Serve the buffered response
+				for k, vv := range coalescedReq.responseHeaders {
 					for _, v := range vv {
 						w.Header().Add(k, v)
 					}
 				}
 				w.Header().Set("X-LanCache-Status", "HIT-COALESCED")
 				w.Header().Set("X-LanCache-Processed-By", "SteamCache2")
-				w.WriteHeader(freshResp.StatusCode)
-				io.Copy(w, freshResp.Body)
+				w.WriteHeader(coalescedReq.statusCode)
+				w.Write(responseData)

 				logger.Logger.Info().
 					Str("key", cacheKey).
@@ -1521,6 +1601,11 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 					coalescedResp.Header[k] = vv
 				}
 				coalescedReq.complete(coalescedResp, nil)
+				// Store the response data for coalesced clients
+				coalescedReq.setResponseData(bodyData)
+
+				// Record cache miss for adaptive/predictive analysis
+				sc.recordCacheMiss(cacheKey, int64(len(bodyData)))
 			}
 		} else {
 			logger.Logger.Warn().
@@ -1556,28 +1641,201 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}

-	// Handle favicon requests
-	if r.URL.Path == "/favicon.ico" {
-		logger.Logger.Debug().
-			Str("client_ip", clientIP).
-			Msg("Favicon request")
-		w.WriteHeader(http.StatusNoContent)
-		return
-	}
-
-	if r.URL.Path == "/robots.txt" {
-		logger.Logger.Debug().
-			Str("client_ip", clientIP).
-			Msg("Robots.txt request")
-		w.Header().Set("Content-Type", "text/plain")
-		w.WriteHeader(http.StatusOK)
-		w.Write([]byte("User-agent: *\nDisallow: /\n"))
-		return
-	}
-
-	logger.Logger.Warn().
-		Str("url", r.URL.String()).
-		Str("client_ip", clientIP).
-		Msg("Request not found")
 	http.Error(w, "Not found", http.StatusNotFound)
 }
+
+// recordCacheAccess records a cache hit for adaptive and predictive analysis (lightweight)
+func (sc *SteamCache) recordCacheAccess(key string, size int64) {
+	// Skip if adaptive features are disabled
+	if !sc.adaptiveEnabled {
+		return
+	}
+
+	// Only record for large files to reduce overhead
+	if size < 1024*1024 { // Skip files smaller than 1MB
+		return
+	}
+
+	// Lightweight adaptive recording
+	sc.adaptiveManager.RecordAccess(key, size)
+
+	// Lightweight predictive recording - only if we have a previous key
+	sc.lastAccessKeyMu.RLock()
+	previousKey := sc.lastAccessKey
+	sc.lastAccessKeyMu.RUnlock()
+
+	if previousKey != "" {
+		sc.predictiveManager.RecordAccess(key, previousKey, size)
+	}
+
+	// Update last accessed key
+	sc.lastAccessKeyMu.Lock()
+	sc.lastAccessKey = key
+	sc.lastAccessKeyMu.Unlock()
+
+	// Skip expensive prefetching on every access
+	// Only do it occasionally to reduce overhead
+}
+
+// recordCacheMiss records a cache miss for adaptive and predictive analysis (lightweight)
+func (sc *SteamCache) recordCacheMiss(key string, size int64) {
+	// Skip if adaptive features are disabled
+	if !sc.adaptiveEnabled {
+		return
+	}
+
+	// Only record for large files to reduce overhead
+	if size < 1024*1024 { // Skip files smaller than 1MB
+		return
+	}
+
+	// Lightweight adaptive recording
+	sc.adaptiveManager.RecordAccess(key, size)
+
+	// Lightweight predictive recording - only if we have a previous key
+	sc.lastAccessKeyMu.RLock()
+	previousKey := sc.lastAccessKey
+	sc.lastAccessKeyMu.RUnlock()
+
+	if previousKey != "" {
+		sc.predictiveManager.RecordAccess(key, previousKey, size)
+	}
+
+	// Update last accessed key
+	sc.lastAccessKeyMu.Lock()
+	sc.lastAccessKey = key
+	sc.lastAccessKeyMu.Unlock()
+
+	// Only trigger warming for very large files to reduce overhead
+	if size > 10*1024*1024 { // Only warm files > 10MB
+		sc.cacheWarmer.RequestWarming(key, 3, "cache_miss", size, "cache_miss_analyzer")
+	}
+}
+
+// GetAdaptiveStats returns adaptive caching statistics
+func (sc *SteamCache) GetAdaptiveStats() map[string]interface{} {
+	stats := make(map[string]interface{})
+
+	// Get current strategy
+	currentStrategy := sc.adaptiveManager.GetCurrentStrategy()
+	stats["current_strategy"] = currentStrategy
+	stats["adaptation_count"] = sc.adaptiveManager.GetAdaptationCount()
+
+	// Get dominant pattern (using public method)
+	// Note: In a real implementation, we'd need a public method to get the dominant pattern
+	stats["dominant_pattern"] = "unknown" // Placeholder for now
+
+	return stats
+}
+
+// GetPredictiveStats returns predictive caching statistics
+func (sc *SteamCache) GetPredictiveStats() map[string]interface{} {
+	stats := make(map[string]interface{})
+
+	predictiveStats := sc.predictiveManager.GetStats()
+	stats["prefetch_hits"] = predictiveStats.PrefetchHits
+	stats["prefetch_misses"] = predictiveStats.PrefetchMisses
+	stats["prefetch_requests"] = predictiveStats.PrefetchRequests
+	stats["cache_warm_hits"] = predictiveStats.CacheWarmHits
+	stats["cache_warm_misses"] = predictiveStats.CacheWarmMisses
+
+	return stats
+}
+
+// GetWarmingStats returns cache warming statistics
+func (sc *SteamCache) GetWarmingStats() map[string]interface{} {
+	stats := make(map[string]interface{})
+
+	warmingStats := sc.cacheWarmer.GetStats()
+	stats["warm_requests"] = warmingStats.WarmRequests
+	stats["warm_successes"] = warmingStats.WarmSuccesses
+	stats["warm_failures"] = warmingStats.WarmFailures
+	stats["warm_bytes"] = warmingStats.WarmBytes
+	stats["warm_duration"] = warmingStats.WarmDuration
+	stats["active_warmers"] = warmingStats.ActiveWarmers
+	stats["warming_enabled"] = sc.cacheWarmer.IsWarmingEnabled()
+
+	return stats
+}
+
+// SetWarmingEnabled enables or disables cache warming
+func (sc *SteamCache) SetWarmingEnabled(enabled bool) {
+	sc.cacheWarmer.SetWarmingEnabled(enabled)
+}
+
+// WarmPopularContent manually triggers warming of popular content
+func (sc *SteamCache) WarmPopularContent(keys []string) {
+	sc.cacheWarmer.WarmPopularContent(keys, 2)
+}
+
+// WarmPredictedContent manually triggers warming of predicted content
+func (sc *SteamCache) WarmPredictedContent(keys []string) {
+	sc.cacheWarmer.WarmPredictedContent(keys, 3)
+}
+
+// SetAdaptiveEnabled enables or disables adaptive features
+func (sc *SteamCache) SetAdaptiveEnabled(enabled bool) {
+	sc.adaptiveEnabled = enabled
+	if !enabled {
+		// Stop adaptive components when disabled
+		sc.adaptiveManager.Stop()
+		sc.predictiveManager.Stop()
+		sc.cacheWarmer.Stop()
+	}
+}
+
+// IsAdaptiveEnabled returns whether adaptive features are enabled
+func (sc *SteamCache) IsAdaptiveEnabled() bool {
+	return sc.adaptiveEnabled
+}
+
+// GetMemoryStats returns memory monitoring statistics
+func (sc *SteamCache) GetMemoryStats() map[string]interface{} {
+	if sc.memoryMonitor == nil {
+		return map[string]interface{}{"error": "memory monitoring not enabled"}
+	}
+
+	stats := sc.memoryMonitor.GetMemoryStats()
+
+	if sc.dynamicCacheMgr != nil {
+		dynamicStats := sc.dynamicCacheMgr.GetStats()
+		for k, v := range dynamicStats {
+			stats["dynamic_"+k] = v
+		}
+	}
+
+	return stats
+}
+
+// GetDynamicCacheStats returns dynamic cache management statistics
+func (sc *SteamCache) GetDynamicCacheStats() map[string]interface{} {
+	if sc.dynamicCacheMgr == nil {
+		return map[string]interface{}{"error": "dynamic cache management not enabled"}
+	}
+
+	return sc.dynamicCacheMgr.GetStats()
+}
+
+// SetMemoryTarget sets the target memory usage for dynamic cache sizing
+func (sc *SteamCache) SetMemoryTarget(targetBytes uint64) {
+	if sc.memoryMonitor != nil {
+		sc.memoryMonitor.SetTargetMemoryUsage(targetBytes)
+	}
+}
+
+// ForceCacheAdjustment forces an immediate cache size adjustment
+func (sc *SteamCache) ForceCacheAdjustment() {
+	if sc.dynamicCacheMgr != nil {
+		// This would trigger an immediate adjustment
+		// Implementation depends on the specific needs
+	}
+}
+
+// GetMemoryFragmentationStats returns memory fragmentation statistics
+func (sc *SteamCache) GetMemoryFragmentationStats() map[string]interface{} {
+	if sc.memory == nil {
+		return map[string]interface{}{"error": "memory cache not enabled"}
+	}
+
+	return sc.memory.GetFragmentationStats()
+}