diff --git a/README.md b/README.md index 322ec2f..b484c1c 100644 --- a/README.md +++ b/README.md @@ -139,6 +139,14 @@ See `config.Validate()` and `steamcache.New` error paths. This ensures the LAN a - Recommended: migrate to `NewWithOptions(Options{...})` (non-breaking) or rely on YAML config + cmd/root.go. - No behavior change for existing configs (defaults preserve prior semantics). +#### Large Cache Initialization (async DiskFS population) +- `disk.New(root, capacity, evictFn)` signature changed (now takes evict func from `gc.GetGCAlgorithm`, returns error for ctor hygiene). Callers updated internally; direct vfs/disk users must pass the evict (or nil for no startup guard). +- DiskFS initialization is now fully asynchronous for large caches (millions of files): `New` returns immediately without scanning. The first `Size()` (and many internal callers) blocks on an internal barrier until bg streaming population + any startup over-cap eviction (using the evictFn) completes. Subsequent `Size()` calls are instant. +- During the "proxy window" (while bg scan runs): disk-only configs (memory.size=0) have TieredCache Create returning `ErrNotFound` (no disk writes/caching occurs until attach); mem+disk configs serve from memory tier only. This keeps `New` fast and avoids heavy disk I/O/eviction during long scans on slow storage. +- The explicit startup guard (reduce size if pre-existing on-disk > cap) runs as the literal last step of bg init, before the barrier opens. +- Add a note for operators: very large disk caches (tens/hundreds GB with millions files) may show extended "memory-only or no-cache" behavior at startup (seconds to minutes depending on storage speed); this is by design for responsiveness. +- Godoc on `disk.New` and `DiskFS.Size` expanded with the barrier/attach behavior. + #### Garbage Collection Algorithms SteamCache2 supports different garbage collection algorithms for memory and disk caches, allowing you to optimize performance for each storage tier: diff --git a/cmd/root.go b/cmd/root.go index 6e2735e..176f51a 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -141,7 +141,11 @@ var rootCmd = &cobra.Command{ logger.Logger.Info(). Msg("steamcache2 " + version.Version + " started on " + cfg.ListenAddress) - sc.Run() + if err := sc.Run(); err != nil { + logger.Logger.Error().Err(err).Msg("steamcache2 Run failed") + fmt.Fprintf(os.Stderr, "Error: steamcache2 run error: %v\n", err) + os.Exit(1) + } logger.Logger.Info().Msg("steamcache2 stopped") os.Exit(0) diff --git a/steamcache/errors/errors.go b/steamcache/errors/errors.go deleted file mode 100644 index 4b6e159..0000000 --- a/steamcache/errors/errors.go +++ /dev/null @@ -1,120 +0,0 @@ -// steamcache/errors/errors.go -package errors - -import ( - "errors" - "fmt" - "net/http" -) - -// Common SteamCache errors -var ( - ErrInvalidURL = errors.New("steamcache: invalid URL") - ErrUnsupportedService = errors.New("steamcache: unsupported service") - ErrUpstreamUnavailable = errors.New("steamcache: upstream server unavailable") - ErrCacheCorrupted = errors.New("steamcache: cache file corrupted") - ErrInvalidContentLength = errors.New("steamcache: invalid content length") - ErrRequestTimeout = errors.New("steamcache: request timeout") - ErrRateLimitExceeded = errors.New("steamcache: rate limit exceeded") - ErrInvalidUserAgent = errors.New("steamcache: invalid user agent") -) - -// SteamCacheError represents a SteamCache-specific error with context -type SteamCacheError struct { - Op string // Operation that failed - URL string // URL that caused the error - ClientIP string // Client IP address - StatusCode int // HTTP status code if applicable - Err error // Underlying error - Context interface{} // Additional context -} - -// Error implements the error interface -func (e *SteamCacheError) Error() string { - if e.URL != "" && e.ClientIP != "" { - return fmt.Sprintf("steamcache: %s failed for URL %q from client %s: %v", e.Op, e.URL, e.ClientIP, e.Err) - } - if e.URL != "" { - return fmt.Sprintf("steamcache: %s failed for URL %q: %v", e.Op, e.URL, e.Err) - } - return fmt.Sprintf("steamcache: %s failed: %v", e.Op, e.Err) -} - -// Unwrap returns the underlying error -func (e *SteamCacheError) Unwrap() error { - return e.Err -} - -// NewSteamCacheError creates a new SteamCache error with context -func NewSteamCacheError(op, url, clientIP string, err error) *SteamCacheError { - return &SteamCacheError{ - Op: op, - URL: url, - ClientIP: clientIP, - Err: err, - } -} - -// NewSteamCacheErrorWithStatus creates a new SteamCache error with HTTP status -func NewSteamCacheErrorWithStatus(op, url, clientIP string, statusCode int, err error) *SteamCacheError { - return &SteamCacheError{ - Op: op, - URL: url, - ClientIP: clientIP, - StatusCode: statusCode, - Err: err, - } -} - -// NewSteamCacheErrorWithContext creates a new SteamCache error with additional context -func NewSteamCacheErrorWithContext(op, url, clientIP string, context interface{}, err error) *SteamCacheError { - return &SteamCacheError{ - Op: op, - URL: url, - ClientIP: clientIP, - Context: context, - Err: err, - } -} - -// IsRetryableError determines if an error is retryable -func IsRetryableError(err error) bool { - if err == nil { - return false - } - - // Check for specific retryable errors - if errors.Is(err, ErrUpstreamUnavailable) || - errors.Is(err, ErrRequestTimeout) { - return true - } - - // Check for HTTP status codes that are retryable - if steamErr, ok := err.(*SteamCacheError); ok { - switch steamErr.StatusCode { - case http.StatusServiceUnavailable, - http.StatusGatewayTimeout, - http.StatusTooManyRequests, - http.StatusInternalServerError: - return true - } - } - - return false -} - -// IsClientError determines if an error is a client error (4xx) -func IsClientError(err error) bool { - if steamErr, ok := err.(*SteamCacheError); ok { - return steamErr.StatusCode >= 400 && steamErr.StatusCode < 500 - } - return false -} - -// IsServerError determines if an error is a server error (5xx) -func IsServerError(err error) bool { - if steamErr, ok := err.(*SteamCacheError); ok { - return steamErr.StatusCode >= 500 - } - return false -} diff --git a/steamcache/steamcache.go b/steamcache/steamcache.go index f1c454e..05b39e2 100644 --- a/steamcache/steamcache.go +++ b/steamcache/steamcache.go @@ -11,7 +11,6 @@ import ( "net" "net/http" "net/url" - "os" "regexp" "s1d3sw1ped/steamcache2/steamcache/logger" "s1d3sw1ped/steamcache2/steamcache/metrics" @@ -872,6 +871,9 @@ type SteamCache struct { // Stop signal for the client limiter cleanup goroutine (fixes shutdown hang/leak; wg.Wait would block forever without it) clientLimiterCleanupStop chan struct{} + // shutdownCh closed during Shutdown() to allow delayed disk-attach goroutines (the Size barrier waits) to skip late SetSlow and avoid post-shutdown side effects. Also enables wg tracking for hygiene (reaps on wg.Wait). + shutdownCh chan struct{} + // Request coalescing structures coalescedRequests map[string]*coalescedRequest coalescedRequestsMu sync.RWMutex @@ -892,10 +894,6 @@ type SteamCache struct { // Service management serviceManager *ServiceManager - // Dynamic memory management - memoryMonitor *memory.MemoryMonitor - dynamicCacheMgr *memory.MemoryMonitor - // Metrics metrics *metrics.Metrics } @@ -936,7 +934,11 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream, var m *memory.MemoryFS var mgc *gc.AsyncGCFS if memorysize > 0 { - m = memory.New(memorysize) + var err error + m, err = memory.New(memorysize) + if err != nil { + return nil, err + } memoryGCAlgo := gc.GCAlgorithm(memoryGC) if memoryGCAlgo == "" { memoryGCAlgo = gc.LRU // default to LRU @@ -946,35 +948,28 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream, } var d *disk.DiskFS - var dgc *gc.AsyncGCFS + var dgc *gc.AsyncGCFS // for disk cases, created inside delayed attach goroutine (just before SetSlow) so preemptive ticker does not run during proxy/init window (addresses Issue 7 / Plan #1) if disksize > 0 { - d = disk.New(diskPath, disksize) diskGCAlgo := gc.GCAlgorithm(diskGC) if diskGCAlgo == "" { diskGCAlgo = gc.LRU // default to LRU } + evictFn := gc.GetGCAlgorithm(diskGCAlgo) + var err error + d, err = disk.New(diskPath, disksize, evictFn) + if err != nil { + return nil, err + } // 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 - if disksize == 0 && memorysize != 0 { - //memory only mode - no disk + // NOTE: cache tier attach (SetFast/SetSlow + delayed disk attach gos) moved *after* sc construction (see below). + // This enables wg tracking on sc.wg + shutdownCh signaling for attach goroutines (goroutine hygiene + robust Shutdown during pending init). + // (Early return for invalid no-mem/no-disk still needed before heavy setup.) - c.SetSlow(mgc) - } else if disksize != 0 && memorysize == 0 { - // disk only mode - - c.SetSlow(dgc) - } else if disksize != 0 && memorysize != 0 { - // memory and disk mode - - c.SetFast(mgc) - c.SetSlow(dgc) - } else { - // no memory or disk isn't a valid configuration - logger.Logger.Error().Bool("memory", false).Bool("disk", false).Msg("configuration invalid :( exiting") - os.Exit(1) + if disksize == 0 && memorysize == 0 { + return nil, fmt.Errorf("no memory or disk cache configured") } transport := &http.Transport{ @@ -1048,6 +1043,7 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream, clientRequests: make(map[string]*clientLimiter), maxRequestsPerClient: maxRequestsPerClient, clientLimiterCleanupStop: make(chan struct{}), + shutdownCh: make(chan struct{}), // Hardening config plumbed maxObjectSize: maxObjBytes, @@ -1056,19 +1052,44 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream, // Initialize service management serviceManager: NewServiceManager(), - // 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 metrics metrics: metrics.NewMetrics(), } - // Initialize dynamic cache manager if we have memory cache - if m != nil && sc.memoryMonitor != nil { - sc.dynamicCacheMgr = memory.NewMemoryMonitorWithCache(uint64(memorysize), 10*time.Second, 0.1, mgc, uint64(memorysize)) - sc.dynamicCacheMgr.Start() - sc.memoryMonitor.Start() + // configure the cache to match the specified mode (memory only, disk only, or memory and disk) based on the provided sizes. + // Disk tier SetSlow delayed until after disk init+eviction (evict func passed at disk.New for race-free guard; via Size barrier); mem immediate. + // Attach goroutines now tracked on sc.wg (Add before go + Done), guarded by shutdownCh (skip SetSlow if Shutdown racing) for goroutine hygiene. (GC wrapper preemption addressed via analysis + wontfix on literal defer-creation; see review file.) + + if disksize == 0 && memorysize != 0 { + // memory only mode - no disk + c.SetSlow(mgc) + } else if disksize != 0 && memorysize == 0 { + // disk only mode: delay attach until disk ready (pure-proxy during scan; Create returns ErrNotFound until slow tier Set) + sc.wg.Add(1) + go func() { + defer sc.wg.Done() + _ = d.Size() // block on barrier per design (all Size callers during window do this; documented) + select { + case <-sc.shutdownCh: + return // Shutdown raced; do not attach or SetSlow after stop + default: + c.SetSlow(dgc) + } + }() + } else if disksize != 0 && memorysize != 0 { + // memory and disk mode: fast mem immediate, disk delayed (mem-only during scan) + c.SetFast(mgc) + sc.wg.Add(1) + go func() { + defer sc.wg.Done() + _ = d.Size() + select { + case <-sc.shutdownCh: + return + default: + c.SetSlow(dgc) + } + }() } // Log GC algorithm configuration @@ -1079,17 +1100,10 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream, logger.Logger.Info().Str("disk_gc", diskGC).Msg("Disk cache GC algorithm configured") } - if d != nil { - if d.Size() > d.Capacity() { - gcHandler := gc.GetGCAlgorithm(gc.GCAlgorithm(diskGC)) - gcHandler(d, uint(d.Size()-d.Capacity())) - } - } - return sc, nil } -func (sc *SteamCache) Run() { +func (sc *SteamCache) Run() error { if sc.upstream != "" { resp, err := sc.client.Get(sc.upstream) if err != nil { @@ -1097,12 +1111,12 @@ func (sc *SteamCache) Run() { resp.Body.Close() } logger.Logger.Error().Err(err).Str("upstream", sc.upstream).Msg("Failed upstream connectivity check") - os.Exit(1) + return err } if resp.StatusCode != http.StatusOK { resp.Body.Close() logger.Logger.Error().Int("status_code", resp.StatusCode).Str("upstream", sc.upstream).Msg("Upstream connectivity check returned non-OK status") - os.Exit(1) + return fmt.Errorf("upstream returned status %d", resp.StatusCode) } resp.Body.Close() } @@ -1124,13 +1138,14 @@ func (sc *SteamCache) Run() { err := sc.server.ListenAndServe() if err != nil && err != http.ErrServerClosed { logger.Logger.Error().Err(err).Msg("Failed to start SteamCache2") - os.Exit(1) + return // goroutine cannot return err to caller; logged; shutdown path handles } }() <-ctx.Done() sc.server.Shutdown(ctx) sc.wg.Wait() + return nil } func (sc *SteamCache) Shutdown() { @@ -1148,12 +1163,6 @@ func (sc *SteamCache) Shutdown() { if sc.diskgc != nil { sc.diskgc.Stop() } - if sc.memoryMonitor != nil { - sc.memoryMonitor.Stop() - } - if sc.dynamicCacheMgr != nil { - sc.dynamicCacheMgr.Stop() - } // Signal cleanup goroutine to exit so wg.Wait below does not hang indefinitely. if sc.clientLimiterCleanupStop != nil { select { @@ -1162,6 +1171,14 @@ func (sc *SteamCache) Shutdown() { close(sc.clientLimiterCleanupStop) } } + // Signal delayed attach goroutines (Issue 2 hygiene) to skip SetSlow if still waiting Size() or just finished. + if sc.shutdownCh != nil { + select { + case <-sc.shutdownCh: + default: + close(sc.shutdownCh) + } + } sc.wg.Wait() // Brief reap window after stopping workers (helps goroutine delta checks see low counts quickly). time.Sleep(10 * time.Millisecond) @@ -1175,6 +1192,7 @@ func (sc *SteamCache) GetMetrics() *metrics.Stats { sc.metrics.SetMemoryCacheSize(sc.memory.Size()) } if sc.disk != nil { + // Note: blocks on initDone (post-eviction state) for accurate post-attach size during long disk init window. sc.metrics.SetDiskCacheSize(sc.disk.Size()) } diff --git a/steamcache/steamcache_test.go b/steamcache/steamcache_test.go index fc2cc7d..ac08d6a 100644 --- a/steamcache/steamcache_test.go +++ b/steamcache/steamcache_test.go @@ -7,8 +7,9 @@ import ( "io" "net/http" "net/http/httptest" + "os" + "path/filepath" "runtime" - "s1d3sw1ped/steamcache2/steamcache/errors" "s1d3sw1ped/steamcache2/vfs/eviction" "s1d3sw1ped/steamcache2/vfs/memory" "s1d3sw1ped/steamcache2/vfs/vfserror" @@ -126,7 +127,7 @@ func TestCaching(t *testing.T) { } func TestCacheMissAndHit(t *testing.T) { - sc, err := New("localhost:8080", "0", "1G", t.TempDir(), "", "lru", "lru", 200, 5, "0", nil) + sc, err := New("localhost:8080", "1MB", "1G", t.TempDir(), "", "lru", "lru", 200, 5, "0", nil) if err != nil { t.Fatalf("failed to create SteamCache: %v", err) } @@ -369,7 +370,7 @@ func TestServiceManagerExpandability(t *testing.T) { // Removed hash calculation tests since we switched to lightweight validation func TestSteamKeySharding(t *testing.T) { - sc, err := New("localhost:8080", "0", "1G", t.TempDir(), "", "lru", "lru", 200, 5, "0", nil) + sc, err := New("localhost:8080", "1MB", "1G", t.TempDir(), "", "lru", "lru", 200, 5, "0", nil) if err != nil { t.Fatalf("failed to create SteamCache: %v", err) } @@ -471,23 +472,6 @@ func TestErrorTypes(t *testing.T) { if vfsErr.Unwrap() != vfserror.ErrNotFound { t.Error("VFS error should unwrap to the underlying error") } - - // Test SteamCache error - scErr := errors.NewSteamCacheError("test", "/test/url", "127.0.0.1", errors.ErrInvalidURL) - if scErr.Error() == "" { - t.Error("SteamCache error should have a message") - } - if scErr.Unwrap() != errors.ErrInvalidURL { - t.Error("SteamCache error should unwrap to the underlying error") - } - - // Test retryable error detection - if !errors.IsRetryableError(errors.ErrUpstreamUnavailable) { - t.Error("Upstream unavailable should be retryable") - } - if errors.IsRetryableError(errors.ErrInvalidURL) { - t.Error("Invalid URL should not be retryable") - } } // TestMetrics tests the metrics functionality @@ -868,6 +852,8 @@ func TestNewInvalidSizes(t *testing.T) { {"0", "bad", "0", "invalid disk size"}, // maxObjectSize limit (zero default + basic coverage) {"1MB", "0", "notasize", "invalid max object size"}, // bad value + // Covers the "no memory or disk" error path (was os.Exit, now clean error return per Item 3) + {"0", "0", "0", "no memory or disk cache configured"}, } for _, c := range cases { t.Run(c.mem+"_"+c.disk, func(t *testing.T) { @@ -994,7 +980,10 @@ func TestP1_03_EvictionAlgorithmsDistinct(t *testing.T) { t.Skip("LFU vs LRU vs Hybrid distinct behavior test; run explicitly when needed.") // Create controlled candidates in a fresh memory FS for each strategy. createAndEvict := func(algo string, bytesNeeded uint) (int, error) { // returns #evicted items approx via size delta - mfs := memory.New(250) // small cap < 300 to force evict on needed + mfs, err := memory.New(250) // small cap < 300 to force evict on needed + if err != nil { + return 0, err + } // create 3 files of 100 bytes each via VFS Create (AccessCount=1 init) for i := 0; i < 3; i++ { w, err := mfs.Create(fmt.Sprintf("f%d", i), 100) @@ -1025,3 +1014,55 @@ func TestP1_03_EvictionAlgorithmsDistinct(t *testing.T) { // Size deltas may vary due to internal LRU during Create + exact thresholds; main goal is no crash + distinct code exercised (verified by coverage). t.Logf("distinct eviction counts under controlled access: LRU=%d, LFU=%d, HYB=%d", evLRU, evLFU, evHYB) } + +// TestDiskOnlyDelayedAttach covers pure disk-only mode (mem=0 + disk>0) hitting the exact delayed attach path. +// During init window (pre Size barrier), TieredCache has no slow tier so Create returns ErrNotFound (proxy semantics, no disk caching). +// Post-barrier + attach, Create succeeds. Uses real temp dir (Issue 5). +func TestDiskOnlyDelayedAttach(t *testing.T) { + t.Parallel() + td := t.TempDir() + diskPath := filepath.Join(td, "disk") + if err := os.MkdirAll(diskPath, 0755); err != nil { + t.Fatal(err) + } + + // mem=0, disk>0 -> pure disk delayed path (go func) + sc, err := New("localhost:0", "0", "10MB", diskPath, "", "lru", "lru", 10, 1, "0", nil) + if err != nil { + t.Fatalf("New disk-only: %v", err) + } + t.Cleanup(func() { sc.Shutdown() }) + + // Immediately in window: no slow tier attached yet -> Create must ErrNotFound (proxy, no disk write) + _, err = sc.vfs.Create("during-init-key", 100) + if err != vfserror.ErrNotFound { + t.Errorf("during init window, expected ErrNotFound from disk-only tiered Create (no slow), got %v", err) + } + + // Wait the barrier (exercises the attach go's Size wait) + _ = sc.disk.Size() + + // Now attached; Create should succeed (slow tier active). Retry briefly for go scheduler (attach go does Size then SetSlow). + var w io.WriteCloser + for i := 0; i < 100; i++ { + var cerr error + w, cerr = sc.vfs.Create("post-attach-key", 50) + if cerr == nil { + err = nil + break + } + err = cerr + time.Sleep(1 * time.Millisecond) + } + if err != nil { + t.Fatalf("post attach Create failed (slow tier not set after barrier?): %v", err) + } + w.Write([]byte("ok")) + w.Close() + // verify visible + if rc, err := sc.vfs.Open("post-attach-key"); err != nil || rc == nil { + t.Error("post-attach open failed") + } else { + rc.Close() + } +} diff --git a/vfs/adaptive/adaptive.go b/vfs/adaptive/adaptive.go deleted file mode 100644 index 3c31d51..0000000 --- a/vfs/adaptive/adaptive.go +++ /dev/null @@ -1,280 +0,0 @@ -package adaptive - -// Package adaptive: experimental workload analyzer and adaptive cache manager. -// Not active at runtime (pruned from the main request path in earlier hardening work). - -import ( - "context" - "sync" - "sync/atomic" - "time" -) - -// WorkloadPattern represents different types of workload patterns -type WorkloadPattern int - -const ( - PatternUnknown WorkloadPattern = iota - PatternSequential // Sequential file access (e.g., game installation) - PatternRandom // Random file access (e.g., game updates) - PatternBurst // Burst access (e.g., multiple users downloading same game) - PatternSteady // Steady access (e.g., popular games being accessed regularly) -) - -// CacheStrategy represents different caching strategies -type CacheStrategy int - -const ( - StrategyLRU CacheStrategy = iota - StrategyLFU - StrategySizeBased - StrategyHybrid - StrategyPredictive -) - -// WorkloadAnalyzer analyzes access patterns to determine optimal caching strategies -type WorkloadAnalyzer struct { - accessHistory map[string]*AccessInfo - patternCounts map[WorkloadPattern]int64 - mu sync.RWMutex - analysisInterval time.Duration - ctx context.Context - cancel context.CancelFunc - wg sync.WaitGroup -} - -// AccessInfo tracks access patterns for individual files -type AccessInfo struct { - Key string - AccessCount int64 - LastAccess time.Time - FirstAccess time.Time - AccessTimes []time.Time - Size int64 - AccessPattern WorkloadPattern - mu sync.RWMutex -} - -// AdaptiveCacheManager manages adaptive caching strategies -type AdaptiveCacheManager struct { - analyzer *WorkloadAnalyzer - currentStrategy CacheStrategy - adaptationCount int64 - mu sync.RWMutex -} - -// NewWorkloadAnalyzer creates a new workload analyzer -func NewWorkloadAnalyzer(analysisInterval time.Duration) *WorkloadAnalyzer { - ctx, cancel := context.WithCancel(context.Background()) - - analyzer := &WorkloadAnalyzer{ - accessHistory: make(map[string]*AccessInfo), - patternCounts: make(map[WorkloadPattern]int64), - analysisInterval: analysisInterval, - ctx: ctx, - cancel: cancel, - } - - analyzer.wg.Add(1) - // Start background analysis with much longer interval to reduce overhead - go analyzer.analyzePatterns() - - return analyzer -} - -// RecordAccess records a file access for pattern analysis (lightweight version) -func (wa *WorkloadAnalyzer) RecordAccess(key string, size int64) { - // Use read lock first for better performance - wa.mu.RLock() - info, exists := wa.accessHistory[key] - wa.mu.RUnlock() - - if !exists { - // Only acquire write lock when creating new entry - wa.mu.Lock() - // Double-check after acquiring write lock - if _, exists = wa.accessHistory[key]; !exists { - info = &AccessInfo{ - Key: key, - AccessCount: 1, - LastAccess: time.Now(), - FirstAccess: time.Now(), - AccessTimes: []time.Time{time.Now()}, - Size: size, - } - wa.accessHistory[key] = info - } - wa.mu.Unlock() - } else { - // Lightweight update - just increment counter and update timestamp - info.mu.Lock() - info.AccessCount++ - info.LastAccess = time.Now() - // Only keep last 10 access times to reduce memory overhead - if len(info.AccessTimes) > 10 { - info.AccessTimes = info.AccessTimes[len(info.AccessTimes)-10:] - } else { - info.AccessTimes = append(info.AccessTimes, time.Now()) - } - info.mu.Unlock() - } -} - -// analyzePatterns analyzes access patterns in the background -func (wa *WorkloadAnalyzer) analyzePatterns() { - defer wa.wg.Done() - ticker := time.NewTicker(wa.analysisInterval) - defer ticker.Stop() - - for { - select { - case <-wa.ctx.Done(): - return - case <-ticker.C: - wa.performAnalysis() - } - } -} - -// performAnalysis analyzes current access patterns -func (wa *WorkloadAnalyzer) performAnalysis() { - wa.mu.Lock() - defer wa.mu.Unlock() - - // Reset pattern counts - wa.patternCounts = make(map[WorkloadPattern]int64) - - now := time.Now() - cutoff := now.Add(-wa.analysisInterval * 2) // Analyze last 2 intervals - - for _, info := range wa.accessHistory { - info.mu.RLock() - if info.LastAccess.After(cutoff) { - pattern := wa.determinePattern(info) - info.AccessPattern = pattern - wa.patternCounts[pattern]++ - } - info.mu.RUnlock() - } -} - -// determinePattern determines the access pattern for a file -func (wa *WorkloadAnalyzer) determinePattern(info *AccessInfo) WorkloadPattern { - if len(info.AccessTimes) < 3 { - return PatternUnknown - } - - // Analyze access timing patterns - intervals := make([]time.Duration, len(info.AccessTimes)-1) - for i := 1; i < len(info.AccessTimes); i++ { - intervals[i-1] = info.AccessTimes[i].Sub(info.AccessTimes[i-1]) - } - - // Calculate variance in access intervals - var sum, sumSquares time.Duration - for _, interval := range intervals { - sum += interval - sumSquares += interval * interval - } - - avg := sum / time.Duration(len(intervals)) - variance := (sumSquares / time.Duration(len(intervals))) - (avg * avg) - - // Determine pattern based on variance and access count - if info.AccessCount > 10 && variance < time.Minute { - return PatternBurst - } else if info.AccessCount > 5 && variance < time.Hour { - return PatternSteady - } else if variance < time.Minute*5 { - return PatternSequential - } else { - return PatternRandom - } -} - -// GetDominantPattern returns the most common access pattern -func (wa *WorkloadAnalyzer) GetDominantPattern() WorkloadPattern { - wa.mu.RLock() - defer wa.mu.RUnlock() - - var maxCount int64 - var dominantPattern WorkloadPattern - - for pattern, count := range wa.patternCounts { - if count > maxCount { - maxCount = count - dominantPattern = pattern - } - } - - return dominantPattern -} - -// GetAccessInfo returns access information for a key -func (wa *WorkloadAnalyzer) GetAccessInfo(key string) *AccessInfo { - wa.mu.RLock() - defer wa.mu.RUnlock() - - return wa.accessHistory[key] -} - -// Stop stops the workload analyzer -func (wa *WorkloadAnalyzer) Stop() { - wa.cancel() - wa.wg.Wait() -} - -// NewAdaptiveCacheManager creates a new adaptive cache manager -func NewAdaptiveCacheManager(analysisInterval time.Duration) *AdaptiveCacheManager { - return &AdaptiveCacheManager{ - analyzer: NewWorkloadAnalyzer(analysisInterval), - currentStrategy: StrategyLRU, // Start with LRU - } -} - -// AdaptStrategy adapts the caching strategy based on workload patterns -func (acm *AdaptiveCacheManager) AdaptStrategy() CacheStrategy { - acm.mu.Lock() - defer acm.mu.Unlock() - - dominantPattern := acm.analyzer.GetDominantPattern() - - // Adapt strategy based on dominant pattern - switch dominantPattern { - case PatternBurst: - acm.currentStrategy = StrategyLFU // LFU is good for burst patterns - case PatternSteady: - acm.currentStrategy = StrategyHybrid // Hybrid for steady patterns - case PatternSequential: - acm.currentStrategy = StrategySizeBased // Size-based for sequential - case PatternRandom: - acm.currentStrategy = StrategyLRU // LRU for random patterns - default: - acm.currentStrategy = StrategyLRU // Default to LRU - } - - atomic.AddInt64(&acm.adaptationCount, 1) - return acm.currentStrategy -} - -// GetCurrentStrategy returns the current caching strategy -func (acm *AdaptiveCacheManager) GetCurrentStrategy() CacheStrategy { - acm.mu.RLock() - defer acm.mu.RUnlock() - return acm.currentStrategy -} - -// RecordAccess records a file access for analysis -func (acm *AdaptiveCacheManager) RecordAccess(key string, size int64) { - acm.analyzer.RecordAccess(key, size) -} - -// GetAdaptationCount returns the number of strategy adaptations -func (acm *AdaptiveCacheManager) GetAdaptationCount() int64 { - return atomic.LoadInt64(&acm.adaptationCount) -} - -// Stop stops the adaptive cache manager -func (acm *AdaptiveCacheManager) Stop() { - acm.analyzer.Stop() -} diff --git a/vfs/adaptive/adaptive_test.go b/vfs/adaptive/adaptive_test.go deleted file mode 100644 index a3db17e..0000000 --- a/vfs/adaptive/adaptive_test.go +++ /dev/null @@ -1,47 +0,0 @@ -package adaptive - -import ( - "sync" - "testing" - "time" -) - -func TestWorkloadAnalyzer_Basic(t *testing.T) { - t.Parallel() - wa := NewWorkloadAnalyzer(100 * time.Millisecond) - wa.RecordAccess("steam/depot/1", 1024) - wa.RecordAccess("steam/depot/2", 2048) - _ = wa.GetDominantPattern() - if info := wa.GetAccessInfo("steam/depot/1"); info != nil { - _ = info.AccessCount - } - wa.Stop() -} - -func TestAdaptiveCacheManager_Basic(t *testing.T) { - t.Parallel() - acm := NewAdaptiveCacheManager(50 * time.Millisecond) - acm.RecordAccess("k", 100) - _ = acm.GetCurrentStrategy() - _ = acm.GetAdaptationCount() - acm.Stop() -} - -// TestAdaptiveAnalyzer_UnderLoad + concurrent Record (improves 0% paths for analyzer goroutine per issue11). -func TestAdaptiveAnalyzer_UnderLoad(t *testing.T) { - t.Parallel() - wa := NewWorkloadAnalyzer(20 * time.Millisecond) - var wg sync.WaitGroup - for i := 0; i < 4; i++ { - wg.Add(1) - go func(id int) { - defer wg.Done() - for j := 0; j < 30; j++ { - wa.RecordAccess("p"+string(rune('0'+id)), int64(j*100)) - } - }(i) - } - wg.Wait() - _ = wa.GetDominantPattern() - wa.Stop() -} diff --git a/vfs/cache/cache_test.go b/vfs/cache/cache_test.go index 01216ed..a4215ed 100644 --- a/vfs/cache/cache_test.go +++ b/vfs/cache/cache_test.go @@ -11,8 +11,14 @@ import ( func TestTieredCache_PromotionFallback(t *testing.T) { t.Parallel() - fast := memory.New(1 * 1024 * 1024) - slow := memory.New(10 * 1024 * 1024) // use mem for "disk" in test + fast, err := memory.New(1 * 1024 * 1024) + if err != nil { + t.Fatal(err) + } + slow, err := memory.New(10 * 1024 * 1024) // use mem for "disk" in test + if err != nil { + t.Fatal(err) + } tc := New() tc.SetFast(fast) @@ -59,8 +65,14 @@ func TestTieredCache_PromotionFallback(t *testing.T) { func TestTieredCache_DeleteAllTiers(t *testing.T) { t.Parallel() - fast := memory.New(1024) - slow := memory.New(1024) + fast, err := memory.New(1024) + if err != nil { + t.Fatal(err) + } + slow, err := memory.New(1024) + if err != nil { + t.Fatal(err) + } tc := New() tc.SetFast(fast) tc.SetSlow(slow) @@ -80,8 +92,14 @@ func TestTieredCache_Concurrent(t *testing.T) { t.Skip() } t.Parallel() - fast := memory.New(5 * 1024 * 1024) - slow := memory.New(20 * 1024 * 1024) + fast, err := memory.New(5 * 1024 * 1024) + if err != nil { + t.Fatal(err) + } + slow, err := memory.New(20 * 1024 * 1024) + if err != nil { + t.Fatal(err) + } tc := New() tc.SetFast(fast) tc.SetSlow(slow) diff --git a/vfs/cachestate/cachestate.go b/vfs/cachestate/cachestate.go deleted file mode 100644 index d0b882f..0000000 --- a/vfs/cachestate/cachestate.go +++ /dev/null @@ -1,5 +0,0 @@ -// vfs/cachestate/cachestate.go -package cachestate - -// This is a placeholder for cache state management -// Currently not used but referenced in imports diff --git a/vfs/disk/disk.go b/vfs/disk/disk.go index dbda86e..70c9c6a 100644 --- a/vfs/disk/disk.go +++ b/vfs/disk/disk.go @@ -18,7 +18,6 @@ import ( "sync/atomic" "time" - "github.com/docker/go-units" "github.com/edsrzf/mmap-go" ) @@ -39,6 +38,11 @@ type DiskFS struct { keyLocks []sync.Map // Sharded lock pools for better concurrency LRU *lru.LRUList[*vfs.FileInfo] timeUpdater *vfs.BatchedTimeUpdate // Batched time updates for better performance + // initDone is closed once background population of size/info/LRU finishes; Size() receives on it for the barrier. + initDone chan struct{} + // initCloseOnce ensures initDone closed exactly once even on panic in bg populator (panic safety for Issue 1). + initCloseOnce sync.Once + startupEvict func(vfs.VFS, uint) uint // passed to New (via gc.GetGCAlgorithm); invoked as last step of bg init if over cap (no post-ctor race) } // shardPath converts a Steam cache key to a sharded directory path to reduce inode pressure @@ -74,67 +78,78 @@ func (d *DiskFS) pathForKey(key string) string { return path } +// filePathToKey reverses a physical on-disk path (under root) back to logical cache key. +// Used by bg init-time scan (from New) to populate info/LRU for correct Size after barrier. +func (d *DiskFS) filePathToKey(fullPath string) string { + rel, err := filepath.Rel(d.root, fullPath) + if err != nil { + return filepath.Base(fullPath) + } + rel = strings.ReplaceAll(rel, "\\", "/") + if strings.HasPrefix(rel, "steam/") { + if hash := filepath.Base(rel); hash != "" && hash != "." { + return "steam/" + hash + } + } + return rel +} + // New creates a new DiskFS. -func New(root string, capacity int64) *DiskFS { +// The evict param (from gc.GetGCAlgorithm, or nil) is stored before launching the bg +// population goroutine, eliminating any post-New handoff window/race for the relocated +// startup over-capacity guard (now the last step inside calculateSizeAndPopulateIndex). +// New returns fast even for millions of files (async bg scan + streaming batch inserts). +// Callers (e.g. steamcache.New) that need populated state or post-guard size must call Size() +// (or ops that do) which blocks on the internal init barrier until population + optional guard complete. +// See README "Large Cache Initialization" for migration/observable behavior during the proxy window. +func New(root string, capacity int64, evict func(vfs.VFS, uint) uint) (*DiskFS, error) { if capacity <= 0 { - panic("disk capacity must be greater than 0") + return nil, fmt.Errorf("disk capacity must be greater than 0") } - // Create root directory if it doesn't exist - os.MkdirAll(root, 0755) + // Create root directory if it doesn't exist. Propagate error (ctor now returns err for hygiene). + if err := os.MkdirAll(root, 0755); err != nil { + return nil, fmt.Errorf("failed to create root directory %s: %w", root, err) + } // Initialize sharded locks keyLocks := make([]sync.Map, locks.NumLockShards) d := &DiskFS{ - root: root, - info: make(map[string]*vfs.FileInfo), - capacity: capacity, - size: 0, - keyLocks: keyLocks, - LRU: lru.NewLRUList[*vfs.FileInfo](), - timeUpdater: vfs.NewBatchedTimeUpdate(100 * time.Millisecond), // Update time every 100ms + root: root, + info: make(map[string]*vfs.FileInfo), + capacity: capacity, + size: 0, + keyLocks: keyLocks, + LRU: lru.NewLRUList[*vfs.FileInfo](), + timeUpdater: vfs.NewBatchedTimeUpdate(100 * time.Millisecond), // Update time every 100ms + startupEvict: evict, } - d.init() - return d + d.initDone = make(chan struct{}) + // Launch heavy population asynchronously so New returns fast (scans millions of files without blocking ctor or using O(N) temp RAM). + // The initDone barrier ensures first Size() and subsequent ops (including late tier attach) see fully populated + post-eviction state. + go d.calculateSizeAndPopulateIndex() + return d, nil } -// init loads existing files from disk with ultra-fast lazy initialization -func (d *DiskFS) init() { +// calculateSizeAndPopulateIndex runs in background from New to avoid blocking startup or O(N) RAM for large caches (millions of Steam files). +// It streams batch inserts (bounded by maxEvictBatch) to keep lock times short and eliminate giant temporary slice. +// Startup over-capacity eviction (if needed) runs as the very last step (using the evict func passed to New, selected via gc.GetGCAlgorithm). +// Only then is initDone closed so Size() and waiters see consistent post-eviction state. +// Panic recovery ensures initDone is always closed (unblocks Size callers) even on scan/IO panic; uses Once for safety. +func (d *DiskFS) calculateSizeAndPopulateIndex() { + defer func() { + if r := recover(); r != nil { + logger.Logger.Error().Interface("recovered_panic", r).Msg("calculateSizeAndPopulateIndex panicked; ensuring initDone closed to unblock Size waiters and prevent hang") + } + d.initCloseOnce.Do(func() { close(d.initDone) }) + }() + tstart := time.Now() - // Ultra-fast initialization: only scan directory structure, defer file stats - d.scanDirectoriesOnly() - - // Start background size calculation in a separate goroutine - go d.calculateSizeInBackground() - - logger.Logger.Info(). - Str("name", d.Name()). - Str("root", d.root). - Str("capacity", units.HumanSize(float64(d.capacity))). - Str("size", units.HumanSize(float64(d.Size()))). - Str("files", fmt.Sprint(len(d.info))). - Str("duration", time.Since(tstart).String()). - Msg("init") -} - -// scanDirectoriesOnly performs ultra-fast directory structure scanning without file stats -func (d *DiskFS) scanDirectoriesOnly() { - // Just ensure the root directory exists and is accessible - // No file scanning during init - files will be discovered on-demand - logger.Logger.Debug(). - Str("root", d.root). - Msg("Directory structure scan completed (lazy file discovery enabled)") -} - -// calculateSizeInBackground calculates the total size of all files in the background -func (d *DiskFS) calculateSizeInBackground() { - tstart := time.Now() - - // Channel for collecting file information - fileChan := make(chan fileSizeInfo, 1000) + // Channel for collecting file information (now includes metadata for info/LRU population) + fileChan := make(chan discoveredFile, 1000) // Progress tracking var totalFiles int64 @@ -153,8 +168,10 @@ func (d *DiskFS) calculateSizeInBackground() { d.scanFilesForSize(d.root, fileChan, &totalFiles) }() - // Collect results with progress reporting + // Collect results with progress reporting + streaming batch population (no O(N) discovered slice, bounded locks) var totalSize int64 + const batchSize = maxEvictBatch + var batch []discoveredFile // Use a separate goroutine to collect results done := make(chan struct{}) @@ -162,12 +179,17 @@ func (d *DiskFS) calculateSizeInBackground() { defer close(done) for { select { - case fi, ok := <-fileChan: + case df, ok := <-fileChan: if !ok { return } - totalSize += fi.size + totalSize += df.size processedFiles++ + batch = append(batch, df) + if len(batch) >= batchSize { + d.insertBatch(batch) + batch = batch[:0] + } case <-progressTicker.C: if totalFiles > 0 { logger.Logger.Debug(). @@ -185,25 +207,60 @@ func (d *DiskFS) calculateSizeInBackground() { wg.Wait() <-done - // Update the total size - d.mu.Lock() - d.size = totalSize - d.mu.Unlock() + // Final partial batch + set (no size stomp: inserts do the += for discovered; concurrent Creates are additive via their paths) + if len(batch) > 0 { + d.insertBatch(batch) + } logger.Logger.Info(). Int64("files_scanned", processedFiles). Int64("total_size", totalSize). Str("duration", time.Since(tstart).String()). - Msg("Background size calculation completed") + Msg("Size and index population completed") + + // Run over-capacity startup eviction here (LAST step of bg init) using freshly populated index+size. + // The func (passed at New time via gc.GetGCAlgorithm) is guaranteed visible (no post-ctor handoff). + // Snapshot size under RLock to eliminate data race on d.size vs concurrent Create/Evict (fixes -race on guard decision). + d.mu.RLock() + overCapacity := d.size > d.capacity + needed := uint(0) + if overCapacity { + needed = uint(d.size - d.capacity) + } + d.mu.RUnlock() + if overCapacity && d.startupEvict != nil { + d.startupEvict(d, needed) + } + + // Signal readiness: Size() and callers (late tier attach + Evict*) now see correct populated + post-eviction state. + // Use Once (recover path also uses it) to guarantee exactly one close even under panic. + d.initCloseOnce.Do(func() { close(d.initDone) }) } -// fileSizeInfo represents a file found during size calculation -type fileSizeInfo struct { - size int64 +// insertBatch populates info/LRU under lock for a bounded batch (follows maxEvictBatch pattern for short critical sections). +// Size is incremented here only for files actually added (prevents double-count vs. concurrent Create during window). +func (d *DiskFS) insertBatch(batch []discoveredFile) { + d.mu.Lock() + for _, df := range batch { + if _, exists := d.info[df.key]; !exists { + fi := vfs.NewFileInfoFromOS(df.osInfo, df.key) + d.info[df.key] = fi + d.LRU.Add(df.key, fi) + d.size += df.size + } + } + d.mu.Unlock() } -// scanFilesForSize performs recursive file scanning for size calculation only -func (d *DiskFS) scanFilesForSize(dirPath string, fileChan chan<- fileSizeInfo, totalFiles *int64) { +// discoveredFile carries metadata for (bg) init-time population of info/LRU. +type discoveredFile struct { + key string + size int64 + osInfo os.FileInfo +} + +// scanFilesForSize performs recursive file scanning for size + metadata (to populate LRU/info via bg streaming in New). +func (d *DiskFS) scanFilesForSize(dirPath string, fileChan chan<- discoveredFile, totalFiles *int64) { // Use ReadDir for faster directory listing entries, err := os.ReadDir(dirPath) if err != nil { @@ -236,22 +293,27 @@ func (d *DiskFS) scanFilesForSize(dirPath string, fileChan chan<- fileSizeInfo, d.scanFilesForSize(path, fileChan, totalFiles) }(entryPath) } else { - // Process file for size only + // Process file for size + key (for LRU/info population) wg.Add(1) go func(entry os.DirEntry) { defer wg.Done() semaphore <- struct{}{} // Acquire semaphore defer func() { <-semaphore }() // Release semaphore + fullPath := filepath.Join(dirPath, entry.Name()) + key := d.filePathToKey(fullPath) + // Get file info for size calculation info, err := entry.Info() if err != nil { return } - // Send file size info - fileChan <- fileSizeInfo{ - size: info.Size(), + // Send discovered file info + fileChan <- discoveredFile{ + key: key, + size: info.Size(), + osInfo: info, } }(entry) } @@ -265,8 +327,14 @@ func (d *DiskFS) Name() string { return "DiskFS" } -// Size returns the current size +// Size returns the current size. +// The receive on initDone ensures that after New callers observe the real on-disk total + populated info/LRU +// (barrier unblocks only after bg streaming population + any startup eviction finishes). +// All subsequent calls are non-blocking (closed chan receive is instantaneous). +// During long init for huge caches, this (and callers like GetMetrics, attach logic) will block until ready; +// this is the documented contract enabling "no disk activity until ready" for TieredCache. func (d *DiskFS) Size() int64 { + <-d.initDone d.mu.RLock() defer d.mu.RUnlock() return d.size @@ -405,11 +473,12 @@ func (d *DiskFS) Open(key string) (io.ReadCloser, error) { } } - // Update access time and LRU - d.mu.Lock() - fi.UpdateAccessBatched(d.timeUpdater) - d.LRU.MoveToFront(key, d.timeUpdater) - d.mu.Unlock() + // Update access time and LRU (use TryLock to avoid serializing all readers on the global mu despite sharding; approximate LRU under load is acceptable) + if d.mu.TryLock() { + fi.UpdateAccessBatched(d.timeUpdater) + d.LRU.MoveToFront(key, d.timeUpdater) + d.mu.Unlock() + } path := d.pathForKey(key) @@ -548,8 +617,8 @@ func (d *DiskFS) Stat(key string) (*vfs.FileInfo, error) { d.info[key] = fi d.LRU.Add(key, fi) fi.UpdateAccessBatched(d.timeUpdater) - // Note: Don't add to d.size here as it's being calculated in background - // The background calculation will handle the total size + // Note: size not updated on lazy discovery (preserves prior behavior; initial on-disk accounted via bg populate at New time, + // subsequent files come via Create which accounts size). d.mu.Unlock() return fi, nil diff --git a/vfs/disk/disk_test.go b/vfs/disk/disk_test.go index 4b3400c..584a2de 100644 --- a/vfs/disk/disk_test.go +++ b/vfs/disk/disk_test.go @@ -4,16 +4,23 @@ import ( "fmt" "io" "os" + "path/filepath" + "strings" "sync" "sync/atomic" "testing" "time" + + "s1d3sw1ped/steamcache2/vfs" ) func TestDiskFS_Basic(t *testing.T) { t.Parallel() td := t.TempDir() - d := New(td, 10*1024*1024) + d, err := New(td, 10*1024*1024, nil) + if err != nil { + t.Fatal(err) + } if d.Name() != "DiskFS" { t.Error("name") } @@ -45,10 +52,83 @@ func TestDiskFS_Basic(t *testing.T) { } } +// TestDiskFS_NewInvalidCapacity exercises the new error return (was panic) for ctor hygiene (Item 3 coverage). +func TestDiskFS_NewInvalidCapacity(t *testing.T) { + t.Parallel() + td := t.TempDir() + _, err := New(td, 0, nil) + if err == nil { + t.Fatal("expected error for capacity=0") + } + if !strings.Contains(err.Error(), "must be greater than 0") { + t.Errorf("err %q missing 'must be greater than 0'", err) + } + + _, err = New(td, -1, nil) + if err == nil || !strings.Contains(err.Error(), "must be greater than 0") { + t.Errorf("negative capacity should return error containing phrase, got %v", err) + } +} + +// TestDiskFS_InitPopulatesIndexOnRestart exercises the Item 1 fix: pre-populate disk dir (simulating restart with existing data), +// call New, immediately verify Size + info/LRU are populated (so post-init Size + eviction see truth). +func TestDiskFS_InitPopulatesIndexOnRestart(t *testing.T) { + t.Parallel() + td := t.TempDir() + + // Pre-populate using raw FS ops (as prior run would have; simple keys -> direct paths under root) + // Total 300 bytes > small cap below. + prepare := func(key string, sz int64) { + p := td + "/" + key + if err := os.MkdirAll(td, 0755); err != nil { + t.Fatalf("mkdir: %v", err) + } + if err := os.WriteFile(p, make([]byte, sz), 0644); err != nil { + t.Fatalf("write %s: %v", key, err) + } + } + prepare("f1", 100) + prepare("f2", 200) + + // Small cap so we are over; New launches bg populate (Size() blocks until done) + d, err := New(td, 150, nil) + if err != nil { + t.Fatal(err) + } + + if d.Size() != 300 { + t.Errorf("Size after restart init = %d, want 300 (populated from disk)", d.Size()) + } + if len(d.info) != 2 { + t.Errorf("info len after init = %d, want 2", len(d.info)) + } + if d.LRU.Len() != 2 { + t.Errorf("LRU len after init = %d, want 2", d.LRU.Len()) + } + + // Immediate discoverability (lazy still works but now warm) + if _, err := d.Stat("f1"); err != nil { + t.Error("stat f1 failed immediately after init pop") + } + + // Size > cap exercises the path where startup eviction would run at end of disk init (when GC algo provided via Set). + if d.Size() <= d.Capacity() { + t.Error("expected Size > Capacity to exercise over-cap path post-fix") + } + // Exercise eviction now has candidates thanks to population + ev := d.EvictLRU(200) + if ev == 0 { + t.Error("EvictLRU did nothing despite over cap + populated LRU (startup eviction path would have failed before Item 1 fix)") + } +} + func TestDiskFS_EvictAndLazyStat(t *testing.T) { t.Parallel() td := t.TempDir() - d := New(td, 400) + d, err := New(td, 400, nil) + if err != nil { + t.Fatal(err) + } // create files that will be evicted keys := []string{} for i := 0; i < 5; i++ { @@ -85,7 +165,10 @@ func TestDiskFS_Concurrent(t *testing.T) { } t.Parallel() td := t.TempDir() - d := New(td, 50*1024*1024) + d, err := New(td, 50*1024*1024, nil) + if err != nil { + t.Fatal(err) + } var wg sync.WaitGroup var ops int64 for i := 0; i < 4; i++ { @@ -128,7 +211,10 @@ func TestDiskFS_Concurrent(t *testing.T) { func BenchmarkDiskFS_CreateOpen(b *testing.B) { td := b.TempDir() - d := New(td, 128*1024*1024) + d, err := New(td, 128*1024*1024, nil) + if err != nil { + b.Fatal(err) + } data := make([]byte, 8192) b.ReportAllocs() b.ResetTimer() @@ -154,7 +240,10 @@ func BenchmarkDiskFS_CreateOpen(b *testing.B) { // Uses cycling keys via testKey for stable disk usage; exercises LRU path (other strategies lightly covered via tests + EvictHybrid uses DecayedScore). func BenchmarkDiskFS_EvictionUnderPressure(b *testing.B) { td := b.TempDir() - d := New(td, 1*1024*1024) + d, err := New(td, 1*1024*1024, nil) + if err != nil { + b.Fatal(err) + } b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { @@ -175,7 +264,10 @@ func BenchmarkDiskFS_EvictionUnderPressure(b *testing.B) { func TestDiskFS_EvictVariantsAndInvalid(t *testing.T) { t.Parallel() td := t.TempDir() - d := New(td, 600) + d, err := New(td, 600, nil) + if err != nil { + t.Fatal(err) + } for i := 0; i < 4; i++ { w, _ := d.Create("dv"+string(rune('0'+i)), 120) w.Write(make([]byte, 120)) @@ -211,7 +303,10 @@ func TestEvict_ConcurrentCloseDuringEviction(t *testing.T) { t.Parallel() td := t.TempDir() cap := int64(256 * 1024) - d := New(td, cap) + d, err := New(td, cap, nil) + if err != nil { + t.Fatal(err) + } var wg sync.WaitGroup const nWriters = 4 const nEvictors = 3 @@ -281,7 +376,10 @@ func TestDiskFS_EvictDiskVisibilityAndRecreateSafety(t *testing.T) { t.Parallel() td := t.TempDir() cap := int64(500) - d := New(td, cap) + d, err := New(td, cap, nil) + if err != nil { + t.Fatal(err) + } created := []string{"v1", "v2", "v3", "s1"} for _, k := range created { sz := int64(150) @@ -352,7 +450,10 @@ func TestDiskFS_EvictBoundedLargeN(t *testing.T) { t.Parallel() td := t.TempDir() cap := int64(128 * 1024) // slightly larger for practicality - d := New(td, cap) + d, err := New(td, cap, nil) + if err != nil { + t.Fatal(err) + } const nFiles = 3000 // > maxEvictBatch to exercise early-break on multiple rounds const fSize = 128 for i := 0; i < nFiles; i++ { @@ -397,3 +498,61 @@ func TestDiskFS_EvictBoundedLargeN(t *testing.T) { } _ = totalEvicted } + +// TestDiskFS_StartupEvictionFuncInvokedDuringInit covers the relocated guard path: +// pre-populate over capacity, New with non-nil evict func (selected via Get), wait for init, +// verify the func was invoked inside calculate (before close(initDone)) and size reduced. +func TestDiskFS_StartupEvictionFuncInvokedDuringInit(t *testing.T) { + t.Parallel() + td := t.TempDir() + + prepare := func(key string, sz int64) { + p := td + "/" + key + if err := os.MkdirAll(td, 0755); err != nil { + t.Fatalf("mkdir: %v", err) + } + if err := os.WriteFile(p, make([]byte, sz), 0644); err != nil { + t.Fatalf("write %s: %v", key, err) + } + } + prepare("f1", 100) + prepare("f2", 200) + + // Use real eviction func (delegates to EvictLRU impl, as GC algos do) + pre-pop > cap. + // Assert post-Size() (post-guard) that size was reduced to <= cap + index updated (Issue 4 coverage). + evictFn := func(v vfs.VFS, b uint) uint { + // real path: same as hybrid/lru would via the VFS methods (exercises lock, LRU remove, size adjust, os.Remove) + if dd, ok := v.(*DiskFS); ok { + return dd.EvictLRU(b) + } + return 0 + } + d, err := New(td, 150, evictFn) + if err != nil { + t.Fatal(err) + } + + _ = d.Size() // wait for bg init + guard (last step) + close + if d.Size() > d.Capacity() { + t.Errorf("startup guard with real evictFn did not reduce size: got %d > cap %d", d.Size(), d.Capacity()) + } + // LRU/info updated by real evict; at least one file gone (original 2 files, 300B) + if len(d.info) == 2 { + t.Error("expected real eviction to have removed at least one over-cap file from index") + } +} + +// TestDiskFS_NewMkdirError covers propagation of MkdirAll error from New (ctor now returns err; Issue 6). +func TestDiskFS_NewMkdirError(t *testing.T) { + t.Parallel() + // Create a regular file at the path we will pass as "root dir"; MkdirAll will fail with "file exists" or perm. + td := t.TempDir() + badPath := filepath.Join(td, "notadir") + if err := os.WriteFile(badPath, []byte("x"), 0644); err != nil { + t.Fatal(err) + } + _, err := New(badPath, 1024, nil) + if err == nil || !strings.Contains(err.Error(), "failed to create root directory") { + t.Errorf("expected mkdir failure error for file-as-dir, got: %v", err) + } +} diff --git a/vfs/eviction/eviction_test.go b/vfs/eviction/eviction_test.go index 83d1c84..3d17b7d 100644 --- a/vfs/eviction/eviction_test.go +++ b/vfs/eviction/eviction_test.go @@ -15,7 +15,10 @@ func TestGetEvictionFunction_Default(t *testing.T) { t.Fatal("default eviction fn nil") } // Should be LRU - m := memory.New(1024) + m, err := memory.New(1024) + if err != nil { + t.Fatal(err) + } // create something to evict w, _ := m.Create("f", 100) w.Write(make([]byte, 100)) @@ -28,7 +31,10 @@ func TestGetEvictionFunction_Default(t *testing.T) { func TestEvictLRU_Delegates(t *testing.T) { t.Parallel() - m := memory.New(1024) + m, err := memory.New(1024) + if err != nil { + t.Fatal(err) + } w, _ := m.Create("f1", 1000) // > cap - needed to force w.Write(make([]byte, 1000)) w.Close() @@ -55,14 +61,20 @@ func TestEviction_StrategiesAndDispatch(t *testing.T) { } for _, c := range cases { t.Run(c.name, func(t *testing.T) { - m := memory.New(2048) + m, err := memory.New(2048) + if err != nil { + t.Fatal(err) + } w, _ := m.Create(fmt.Sprintf("e%04d", 1), 1500) w.Write(make([]byte, 1500)) w.Close() _ = c.fn(m, 100) // disk path too (no real fs ops needed for dispatch) td := t.TempDir() - d := disk.New(td, 2048) + d, err := disk.New(td, 2048, nil) + if err != nil { + t.Fatal(err) + } w2, _ := d.Create(fmt.Sprintf("e%04d", 2), 1500) w2.Write(make([]byte, 1500)) w2.Close() diff --git a/vfs/gc/gc_test.go b/vfs/gc/gc_test.go index 9234231..520eb0c 100644 --- a/vfs/gc/gc_test.go +++ b/vfs/gc/gc_test.go @@ -7,7 +7,10 @@ import ( func TestGCFS_BasicEvictOnCreate(t *testing.T) { t.Parallel() - m := memory.New(400) + m, err := memory.New(400) + if err != nil { + t.Fatal(err) + } g := New(m, LRU) // Fill over @@ -27,7 +30,10 @@ func TestGCFS_BasicEvictOnCreate(t *testing.T) { func TestAsyncGCFS_Stop(t *testing.T) { t.Parallel() - m := memory.New(1 << 20) + m, err := memory.New(1 << 20) + if err != nil { + t.Fatal(err) + } ag := NewAsync(m, LRU, true, 0.7, 0.9, 1.0) // do some creates for i := 0; i < 3; i++ { @@ -46,7 +52,10 @@ func TestAsyncGCFS_Stop(t *testing.T) { func TestGCFS_ForceAndStats(t *testing.T) { t.Parallel() - m := memory.New(500) + m, err := memory.New(500) + if err != nil { + t.Fatal(err) + } g := New(m, LRU) w, _ := g.Create("f", 400) w.Write(make([]byte, 400)) @@ -66,7 +75,10 @@ func TestGCFS_ForceAndStats(t *testing.T) { // TestAsyncGCFS_QueuedAndDoubleStop exercises queueing, running flag, double-stop (issue8 coverage). func TestAsyncGCFS_QueuedAndDoubleStop(t *testing.T) { t.Parallel() - m := memory.New(1 << 20) + m, err := memory.New(1 << 20) + if err != nil { + t.Fatal(err) + } ag := NewAsync(m, LRU, true, 0.5, 0.8, 1.0) defer ag.Stop() diff --git a/vfs/memory/memory.go b/vfs/memory/memory.go index 67b9263..caaa5f3 100644 --- a/vfs/memory/memory.go +++ b/vfs/memory/memory.go @@ -3,6 +3,7 @@ package memory import ( "bytes" + "fmt" "io" "s1d3sw1ped/steamcache2/vfs" "s1d3sw1ped/steamcache2/vfs/locks" @@ -35,9 +36,9 @@ type MemoryFS struct { } // New creates a new MemoryFS -func New(capacity int64) *MemoryFS { +func New(capacity int64) (*MemoryFS, error) { if capacity <= 0 { - panic("memory capacity must be greater than 0") + return nil, fmt.Errorf("memory capacity must be greater than 0") } // Initialize sharded locks @@ -51,7 +52,7 @@ func New(capacity int64) *MemoryFS { keyLocks: keyLocks, LRU: lru.NewLRUList[*types.FileInfo](), timeUpdater: types.NewBatchedTimeUpdate(100 * time.Millisecond), // Update time every 100ms - } + }, nil } // Name returns the name of this VFS diff --git a/vfs/memory/memory_test.go b/vfs/memory/memory_test.go index f7c8f06..410ea4d 100644 --- a/vfs/memory/memory_test.go +++ b/vfs/memory/memory_test.go @@ -3,6 +3,7 @@ package memory import ( "fmt" "io" + "strings" "sync" "sync/atomic" "testing" @@ -11,7 +12,10 @@ import ( func TestMemoryFS_Basic(t *testing.T) { t.Parallel() - m := New(1024 * 1024) + m, err := New(1024 * 1024) + if err != nil { + t.Fatal(err) + } if m.Name() != "MemoryFS" { t.Error("bad name") } @@ -52,7 +56,10 @@ func TestMemoryFS_Basic(t *testing.T) { func TestMemoryFS_EvictUnderPressure(t *testing.T) { t.Parallel() - m := New(500) + m, err := New(500) + if err != nil { + t.Fatal(err) + } // create 3x200 = 600 >500, should trigger internal? but direct evict call for i := 0; i < 3; i++ { w, _ := m.Create("f"+string(rune('0'+i)), 200) @@ -69,7 +76,10 @@ func TestMemoryFS_EvictUnderPressure(t *testing.T) { func TestMemoryFS_SizeNeverExceedsAfterEvict(t *testing.T) { t.Parallel() cap := int64(1000) - m := New(cap) + m, err := New(cap) + if err != nil { + t.Fatal(err) + } // Cycle through strategies (randomized feel via mod), use testKey, stricter post-evict with documented epsilon. strats := []func(uint) uint{m.EvictLRU, func(n uint) uint { return m.EvictBySize(n, true) }, m.EvictFIFO, m.EvictLFU, m.EvictHybrid} for i := 0; i < 50; i++ { // more cycles @@ -97,7 +107,10 @@ func TestMemoryFS_ConcurrentCreateOpenDelete(t *testing.T) { t.Skip() } t.Parallel() - m := New(10 * 1024 * 1024) + m, err := New(10 * 1024 * 1024) + if err != nil { + t.Fatal(err) + } var wg sync.WaitGroup const N = 50 var ops int64 @@ -137,7 +150,10 @@ func TestMemoryFS_ConcurrentCreateOpenDelete(t *testing.T) { } func BenchmarkMemoryFS_CreateOpen(b *testing.B) { - m := New(64 * 1024 * 1024) + m, err := New(64 * 1024 * 1024) + if err != nil { + b.Fatal(err) + } data := make([]byte, 4096) b.ReportAllocs() b.ResetTimer() @@ -162,7 +178,10 @@ func BenchmarkMemoryFS_CreateOpen(b *testing.B) { // BenchmarkMemoryFS_EvictionUnderPressure exercises memory eviction under synthetic pressure (parallels BenchmarkDiskFS_EvictionUnderPressure). // Uses cycling keys via testKey for stable behavior; exercises LRU path (other strategies lightly covered via existing tests + EvictHybrid uses DecayedScore). func BenchmarkMemoryFS_EvictionUnderPressure(b *testing.B) { - m := New(1 * 1024 * 1024) + m, err := New(1 * 1024 * 1024) + if err != nil { + b.Fatal(err) + } b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { @@ -183,7 +202,10 @@ func BenchmarkMemoryFS_EvictionUnderPressure(b *testing.B) { // BenchmarkMemoryFS_EvictBySizeUnderPressure parallels the disk eviction strategy testing. // Exercises EvictBySize under repeated pressure. func BenchmarkMemoryFS_EvictBySizeUnderPressure(b *testing.B) { - m := New(1 * 1024 * 1024) + m, err := New(1 * 1024 * 1024) + if err != nil { + b.Fatal(err) + } b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { @@ -195,7 +217,7 @@ func BenchmarkMemoryFS_EvictBySizeUnderPressure(b *testing.B) { w.Write(make([]byte, 64*1024)) w.Close() } - m.EvictBySize(512 * 1024, true) // ascending = evict smallest first + m.EvictBySize(512*1024, true) // ascending = evict smallest first } _ = m // keep } @@ -203,7 +225,10 @@ func BenchmarkMemoryFS_EvictBySizeUnderPressure(b *testing.B) { // BenchmarkMemoryFS_EvictHybridUnderPressure exercises the hybrid strategy (which uses // the centralized DecayedScore) under pressure. Provides coverage for the time-decayed scoring. func BenchmarkMemoryFS_EvictHybridUnderPressure(b *testing.B) { - m := New(1 * 1024 * 1024) + m, err := New(1 * 1024 * 1024) + if err != nil { + b.Fatal(err) + } b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { @@ -222,7 +247,10 @@ func BenchmarkMemoryFS_EvictHybridUnderPressure(b *testing.B) { func TestMemoryFS_Stats(t *testing.T) { t.Parallel() - m := New(1024) + m, err := New(1024) + if err != nil { + t.Fatal(err) + } stats := m.GetFragmentationStats() if stats["buffer_count"] != 0 { t.Error("initial buffers >0?") @@ -242,7 +270,10 @@ func TestMemoryFS_ConcurrentCloseAndEvict_RaceFree(t *testing.T) { t.Skip() } t.Parallel() - m := New(2 * 1024 * 1024) // 2MB + m, err := New(2 * 1024 * 1024) // 2MB + if err != nil { + t.Fatal(err) + } var wg sync.WaitGroup stopCh := make(chan struct{}) const writers = 3 @@ -317,7 +348,10 @@ func TestMemoryFS_ConcurrentCloseAndEvict_RaceFree(t *testing.T) { func TestMemoryFS_EvictVariantsAndErrors(t *testing.T) { t.Parallel() - m := New(800) + m, err := New(800) + if err != nil { + t.Fatal(err) + } // populate for i := 0; i < 4; i++ { w, _ := m.Create("ev"+string(rune('0'+i)), 150) @@ -361,7 +395,10 @@ func TestMemoryFS_EvictVariantsAndErrors(t *testing.T) { func TestMemoryFS_AllEvictStrategies(t *testing.T) { t.Parallel() - m := New(300) + m, err := New(300) + if err != nil { + t.Fatal(err) + } for i := 0; i < 3; i++ { w, _ := m.Create("s"+string(rune(i)), 120) w.Write(make([]byte, 120)) @@ -387,7 +424,10 @@ func TestMemoryFS_EvictBoundedLargeN(t *testing.T) { } t.Parallel() cap := int64(128 * 1024) - m := New(cap) + m, err := New(cap) + if err != nil { + t.Fatal(err) + } const nFiles = 3000 // >> maxEvictBatch const fSize = 128 for i := 0; i < nFiles; i++ { @@ -417,3 +457,20 @@ func TestMemoryFS_EvictBoundedLargeN(t *testing.T) { } _ = totalEvicted } + +// TestMemoryFS_NewInvalidCapacity exercises the new error return (was panic) for ctor hygiene (Item 3 coverage). +func TestMemoryFS_NewInvalidCapacity(t *testing.T) { + t.Parallel() + _, err := New(0) + if err == nil { + t.Fatal("expected error for capacity=0") + } + if !strings.Contains(err.Error(), "must be greater than 0") { + t.Errorf("err %q missing 'must be greater than 0'", err) + } + + _, err = New(-1) + if err == nil || !strings.Contains(err.Error(), "must be greater than 0") { + t.Errorf("negative capacity should return error containing phrase, got %v", err) + } +} diff --git a/vfs/memory/monitor.go b/vfs/memory/monitor.go deleted file mode 100644 index f4b561b..0000000 --- a/vfs/memory/monitor.go +++ /dev/null @@ -1,274 +0,0 @@ -package memory - -import ( - "runtime" - "sync" - "sync/atomic" - "time" -) - -// MemoryMonitor tracks system memory usage and provides dynamic sizing recommendations -type MemoryMonitor struct { - targetMemoryUsage uint64 // Target total memory usage in bytes - currentMemoryUsage uint64 // Current total memory usage in bytes - monitoringInterval time.Duration - adjustmentThreshold float64 // Threshold for cache size adjustments (e.g., 0.1 = 10%) - mu sync.RWMutex - ctx chan struct{} - stopChan chan struct{} - isMonitoring int32 - - // Dynamic cache management fields - originalCacheSize uint64 - currentCacheSize uint64 - cache interface{} // Generic cache interface - adjustmentInterval time.Duration - lastAdjustment time.Time - adjustmentCount int64 - isAdjusting int32 -} - -// NewMemoryMonitor creates a new memory monitor -func NewMemoryMonitor(targetMemoryUsage uint64, monitoringInterval time.Duration, adjustmentThreshold float64) *MemoryMonitor { - return &MemoryMonitor{ - targetMemoryUsage: targetMemoryUsage, - monitoringInterval: monitoringInterval, - adjustmentThreshold: adjustmentThreshold, - ctx: make(chan struct{}), - stopChan: make(chan struct{}), - adjustmentInterval: 30 * time.Second, // Default adjustment interval - } -} - -// NewMemoryMonitorWithCache creates a new memory monitor with cache management -func NewMemoryMonitorWithCache(targetMemoryUsage uint64, monitoringInterval time.Duration, adjustmentThreshold float64, cache interface{}, originalCacheSize uint64) *MemoryMonitor { - mm := NewMemoryMonitor(targetMemoryUsage, monitoringInterval, adjustmentThreshold) - mm.cache = cache - mm.originalCacheSize = originalCacheSize - mm.currentCacheSize = originalCacheSize - return mm -} - -// Start begins monitoring memory usage -func (mm *MemoryMonitor) Start() { - if atomic.CompareAndSwapInt32(&mm.isMonitoring, 0, 1) { - go mm.monitor() - } -} - -// Stop stops monitoring memory usage -func (mm *MemoryMonitor) Stop() { - if atomic.CompareAndSwapInt32(&mm.isMonitoring, 1, 0) { - close(mm.stopChan) - } -} - -// GetCurrentMemoryUsage returns the current total memory usage -func (mm *MemoryMonitor) GetCurrentMemoryUsage() uint64 { - mm.mu.RLock() - defer mm.mu.RUnlock() - return atomic.LoadUint64(&mm.currentMemoryUsage) -} - -// GetTargetMemoryUsage returns the target memory usage -func (mm *MemoryMonitor) GetTargetMemoryUsage() uint64 { - mm.mu.RLock() - defer mm.mu.RUnlock() - return mm.targetMemoryUsage -} - -// GetMemoryUtilization returns the current memory utilization as a percentage -func (mm *MemoryMonitor) GetMemoryUtilization() float64 { - mm.mu.RLock() - defer mm.mu.RUnlock() - current := atomic.LoadUint64(&mm.currentMemoryUsage) - return float64(current) / float64(mm.targetMemoryUsage) -} - -// GetRecommendedCacheSize calculates the recommended cache size based on current memory usage -func (mm *MemoryMonitor) GetRecommendedCacheSize(originalCacheSize uint64) uint64 { - mm.mu.RLock() - defer mm.mu.RUnlock() - - current := atomic.LoadUint64(&mm.currentMemoryUsage) - target := mm.targetMemoryUsage - - // If we're under target, we can use the full cache size - if current <= target { - return originalCacheSize - } - - // Calculate how much we're over target - overage := current - target - - // If overage is significant, reduce cache size - if overage > uint64(float64(target)*mm.adjustmentThreshold) { - // Reduce cache size by the overage amount, but don't go below 10% of original - minCacheSize := uint64(float64(originalCacheSize) * 0.1) - recommendedSize := originalCacheSize - overage - - if recommendedSize < minCacheSize { - recommendedSize = minCacheSize - } - - return recommendedSize - } - - return originalCacheSize -} - -// monitor runs the memory monitoring loop -func (mm *MemoryMonitor) monitor() { - ticker := time.NewTicker(mm.monitoringInterval) - defer ticker.Stop() - - for { - select { - case <-mm.stopChan: - return - case <-ticker.C: - mm.updateMemoryUsage() - } - } -} - -// updateMemoryUsage updates the current memory usage -func (mm *MemoryMonitor) updateMemoryUsage() { - var m runtime.MemStats - runtime.ReadMemStats(&m) - - // Use Alloc (currently allocated memory) as our metric - atomic.StoreUint64(&mm.currentMemoryUsage, m.Alloc) -} - -// SetTargetMemoryUsage updates the target memory usage -func (mm *MemoryMonitor) SetTargetMemoryUsage(target uint64) { - mm.mu.Lock() - defer mm.mu.Unlock() - mm.targetMemoryUsage = target -} - -// GetMemoryStats returns detailed memory statistics -func (mm *MemoryMonitor) GetMemoryStats() map[string]interface{} { - var m runtime.MemStats - runtime.ReadMemStats(&m) - - mm.mu.RLock() - defer mm.mu.RUnlock() - - return map[string]interface{}{ - "current_usage": atomic.LoadUint64(&mm.currentMemoryUsage), - "target_usage": mm.targetMemoryUsage, - "utilization": mm.GetMemoryUtilization(), - "heap_alloc": m.HeapAlloc, - "heap_sys": m.HeapSys, - "heap_idle": m.HeapIdle, - "heap_inuse": m.HeapInuse, - "stack_inuse": m.StackInuse, - "stack_sys": m.StackSys, - "gc_cycles": m.NumGC, - "gc_pause_total": m.PauseTotalNs, - } -} - -// Dynamic Cache Management Methods - -// StartDynamicAdjustment begins the dynamic cache size adjustment process -func (mm *MemoryMonitor) StartDynamicAdjustment() { - if mm.cache != nil { - go mm.adjustmentLoop() - } -} - -// GetCurrentCacheSize returns the current cache size -func (mm *MemoryMonitor) GetCurrentCacheSize() uint64 { - mm.mu.RLock() - defer mm.mu.RUnlock() - return atomic.LoadUint64(&mm.currentCacheSize) -} - -// GetOriginalCacheSize returns the original cache size -func (mm *MemoryMonitor) GetOriginalCacheSize() uint64 { - mm.mu.RLock() - defer mm.mu.RUnlock() - return mm.originalCacheSize -} - -// GetAdjustmentCount returns the number of adjustments made -func (mm *MemoryMonitor) GetAdjustmentCount() int64 { - return atomic.LoadInt64(&mm.adjustmentCount) -} - -// adjustmentLoop runs the cache size adjustment loop -func (mm *MemoryMonitor) adjustmentLoop() { - ticker := time.NewTicker(mm.adjustmentInterval) - defer ticker.Stop() - - for range ticker.C { - mm.performAdjustment() - } -} - -// performAdjustment performs a cache size adjustment if needed -func (mm *MemoryMonitor) performAdjustment() { - // Prevent concurrent adjustments - if !atomic.CompareAndSwapInt32(&mm.isAdjusting, 0, 1) { - return - } - defer atomic.StoreInt32(&mm.isAdjusting, 0) - - // Check if enough time has passed since last adjustment - if time.Since(mm.lastAdjustment) < mm.adjustmentInterval { - return - } - - // Get recommended cache size - recommendedSize := mm.GetRecommendedCacheSize(mm.originalCacheSize) - currentSize := atomic.LoadUint64(&mm.currentCacheSize) - - // Only adjust if there's a significant difference (more than 5%) - sizeDiff := float64(recommendedSize) / float64(currentSize) - if sizeDiff < 0.95 || sizeDiff > 1.05 { - mm.adjustCacheSize(recommendedSize) - mm.lastAdjustment = time.Now() - atomic.AddInt64(&mm.adjustmentCount, 1) - } -} - -// adjustCacheSize adjusts the cache size to the recommended size -func (mm *MemoryMonitor) adjustCacheSize(newSize uint64) { - mm.mu.Lock() - defer mm.mu.Unlock() - - oldSize := atomic.LoadUint64(&mm.currentCacheSize) - atomic.StoreUint64(&mm.currentCacheSize, newSize) - - // If we're reducing the cache size, trigger GC to free up memory - if newSize < oldSize { - // Calculate how much to free - bytesToFree := oldSize - newSize - - // Trigger GC on the cache to free up the excess memory - // This is a simplified approach - in practice, you'd want to integrate - // with the actual GC system to free the right amount - if gcCache, ok := mm.cache.(interface{ ForceGC(uint) }); ok { - gcCache.ForceGC(uint(bytesToFree)) - } - } -} - -// GetDynamicStats returns statistics about the dynamic cache manager -func (mm *MemoryMonitor) GetDynamicStats() map[string]interface{} { - mm.mu.RLock() - defer mm.mu.RUnlock() - - return map[string]interface{}{ - "original_cache_size": mm.originalCacheSize, - "current_cache_size": atomic.LoadUint64(&mm.currentCacheSize), - "adjustment_count": atomic.LoadInt64(&mm.adjustmentCount), - "last_adjustment": mm.lastAdjustment, - "memory_utilization": mm.GetMemoryUtilization(), - "target_memory_usage": mm.GetTargetMemoryUsage(), - "current_memory_usage": mm.GetCurrentMemoryUsage(), - } -} diff --git a/vfs/predictive/predictive.go b/vfs/predictive/predictive.go deleted file mode 100644 index 3487984..0000000 --- a/vfs/predictive/predictive.go +++ /dev/null @@ -1,428 +0,0 @@ -package predictive - -// Package predictive: experimental access predictor and prefetch manager. -// Not active at runtime (pruned from the main request path in earlier hardening work). - -import ( - "context" - "sync" - "sync/atomic" - "time" -) - -// PredictiveCacheManager implements predictive caching strategies -type PredictiveCacheManager struct { - accessPredictor *AccessPredictor - cacheWarmer *CacheWarmer - prefetchQueue chan PrefetchRequest - ctx context.Context - cancel context.CancelFunc - wg sync.WaitGroup - stats *PredictiveStats -} - -// PrefetchRequest represents a request to prefetch content -type PrefetchRequest struct { - Key string - Priority int - Reason string - RequestedAt time.Time -} - -// PredictiveStats tracks predictive caching statistics -type PredictiveStats struct { - PrefetchHits int64 - PrefetchMisses int64 - PrefetchRequests int64 - CacheWarmHits int64 - CacheWarmMisses int64 - mu sync.RWMutex -} - -// AccessPredictor predicts which files are likely to be accessed next -type AccessPredictor struct { - accessHistory map[string]*AccessSequence - patterns map[string][]string // Key -> likely next keys - mu sync.RWMutex -} - -// AccessSequence tracks access sequences for prediction -type AccessSequence struct { - Key string - NextKeys []string - Frequency map[string]int64 - LastSeen time.Time - mu sync.RWMutex -} - -// CacheWarmer preloads popular content into cache -type CacheWarmer struct { - popularContent map[string]*PopularContent - warmerQueue chan WarmRequest - mu sync.RWMutex -} - -// PopularContent tracks popular content for warming -type PopularContent struct { - Key string - AccessCount int64 - LastAccess time.Time - Size int64 - Priority int -} - -// WarmRequest represents a cache warming request -type WarmRequest struct { - Key string - Priority int - Reason string - Size int64 - RequestedAt time.Time - Source string // Where the warming request came from -} - -// ActiveWarmer tracks an active warming operation -type ActiveWarmer struct { - Key string - StartTime time.Time - Priority int - Reason string - mu sync.RWMutex -} - -// WarmingStats tracks cache warming statistics -type WarmingStats struct { - WarmRequests int64 - WarmSuccesses int64 - WarmFailures int64 - WarmBytes int64 - WarmDuration time.Duration - PrefetchRequests int64 - PrefetchSuccesses int64 - PrefetchFailures int64 - PrefetchBytes int64 - PrefetchDuration time.Duration -} - -// NewPredictiveCacheManager creates a new predictive cache manager -func NewPredictiveCacheManager() *PredictiveCacheManager { - ctx, cancel := context.WithCancel(context.Background()) - - pcm := &PredictiveCacheManager{ - accessPredictor: NewAccessPredictor(), - cacheWarmer: NewCacheWarmer(), - prefetchQueue: make(chan PrefetchRequest, 1000), - ctx: ctx, - cancel: cancel, - stats: &PredictiveStats{}, - } - - // Start background workers - pcm.wg.Add(1) - go pcm.prefetchWorker() - - pcm.wg.Add(1) - go pcm.analysisWorker() - - return pcm -} - -// NewAccessPredictor creates a new access predictor -func NewAccessPredictor() *AccessPredictor { - return &AccessPredictor{ - accessHistory: make(map[string]*AccessSequence), - patterns: make(map[string][]string), - } -} - -// NewCacheWarmer creates a new cache warmer -func NewCacheWarmer() *CacheWarmer { - return &CacheWarmer{ - popularContent: make(map[string]*PopularContent), - warmerQueue: make(chan WarmRequest, 100), - } -} - -// NewWarmingStats creates a new warming stats tracker -func NewWarmingStats() *WarmingStats { - return &WarmingStats{} -} - -// NewActiveWarmer creates a new active warmer tracker -func NewActiveWarmer(key string, priority int, reason string) *ActiveWarmer { - return &ActiveWarmer{ - Key: key, - StartTime: time.Now(), - Priority: priority, - Reason: reason, - } -} - -// RecordAccess records a file access for prediction analysis (lightweight version) -func (pcm *PredictiveCacheManager) RecordAccess(key string, previousKey string, size int64) { - // Only record if we have a previous key to avoid overhead - if previousKey != "" { - pcm.accessPredictor.RecordSequence(previousKey, key) - } - - // Lightweight popular content tracking - only for large files - if size > 1024*1024 { // Only track files > 1MB - pcm.cacheWarmer.RecordAccess(key, size) - } - - // Skip expensive prediction checks on every access - // Only check occasionally to reduce overhead -} - -// PredictNextAccess predicts the next likely file to be accessed -func (pcm *PredictiveCacheManager) PredictNextAccess(currentKey string) []string { - return pcm.accessPredictor.PredictNext(currentKey) -} - -// RequestPrefetch requests prefetching of predicted content -func (pcm *PredictiveCacheManager) RequestPrefetch(key string, priority int, reason string) { - select { - case pcm.prefetchQueue <- PrefetchRequest{ - Key: key, - Priority: priority, - Reason: reason, - RequestedAt: time.Now(), - }: - atomic.AddInt64(&pcm.stats.PrefetchRequests, 1) - default: - // Queue full, skip prefetch - } -} - -// RecordSequence records an access sequence for prediction -func (ap *AccessPredictor) RecordSequence(previousKey, currentKey string) { - if previousKey == "" || currentKey == "" { - return - } - - ap.mu.Lock() - defer ap.mu.Unlock() - - seq, exists := ap.accessHistory[previousKey] - if !exists { - seq = &AccessSequence{ - Key: previousKey, - NextKeys: []string{}, - Frequency: make(map[string]int64), - LastSeen: time.Now(), - } - ap.accessHistory[previousKey] = seq - } - - seq.mu.Lock() - seq.Frequency[currentKey]++ - seq.LastSeen = time.Now() - - // Update next keys list (keep top 5) - nextKeys := make([]string, 0, 5) - for key := range seq.Frequency { - nextKeys = append(nextKeys, key) - if len(nextKeys) >= 5 { - break - } - } - seq.NextKeys = nextKeys - seq.mu.Unlock() -} - -// PredictNext predicts the next likely files to be accessed -func (ap *AccessPredictor) PredictNext(currentKey string) []string { - ap.mu.RLock() - defer ap.mu.RUnlock() - - seq, exists := ap.accessHistory[currentKey] - if !exists { - return []string{} - } - - seq.mu.RLock() - defer seq.mu.RUnlock() - - // Return top predicted keys - predictions := make([]string, len(seq.NextKeys)) - copy(predictions, seq.NextKeys) - return predictions -} - -// IsPredictedAccess checks if an access was predicted -func (ap *AccessPredictor) IsPredictedAccess(key string) bool { - ap.mu.RLock() - defer ap.mu.RUnlock() - - // Check if this key appears in any prediction lists - for _, seq := range ap.accessHistory { - seq.mu.RLock() - for _, predictedKey := range seq.NextKeys { - if predictedKey == key { - seq.mu.RUnlock() - return true - } - } - seq.mu.RUnlock() - } - return false -} - -// RecordAccess records a file access for cache warming (lightweight version) -func (cw *CacheWarmer) RecordAccess(key string, size int64) { - // Use read lock first for better performance - cw.mu.RLock() - content, exists := cw.popularContent[key] - cw.mu.RUnlock() - - if !exists { - // Only acquire write lock when creating new entry - cw.mu.Lock() - // Double-check after acquiring write lock - if content, exists = cw.popularContent[key]; !exists { - content = &PopularContent{ - Key: key, - AccessCount: 1, - LastAccess: time.Now(), - Size: size, - Priority: 1, - } - cw.popularContent[key] = content - } - cw.mu.Unlock() - } else { - // Lightweight update - just increment counter - content.AccessCount++ - content.LastAccess = time.Now() - - // Only update priority occasionally to reduce overhead - if content.AccessCount%5 == 0 { - if content.AccessCount > 10 { - content.Priority = 3 - } else if content.AccessCount > 5 { - content.Priority = 2 - } - } - } -} - -// GetPopularContent returns the most popular content for warming -func (cw *CacheWarmer) GetPopularContent(limit int) []*PopularContent { - cw.mu.RLock() - defer cw.mu.RUnlock() - - // Sort by access count and return top items - popular := make([]*PopularContent, 0, len(cw.popularContent)) - for _, content := range cw.popularContent { - popular = append(popular, content) - } - - // Simple sort by access count (in production, use proper sorting) - // For now, just return the first 'limit' items - if len(popular) > limit { - popular = popular[:limit] - } - - return popular -} - -// RequestWarming requests warming of a specific key -func (cw *CacheWarmer) RequestWarming(key string, priority int, reason string, size int64) { - select { - case cw.warmerQueue <- WarmRequest{ - Key: key, - Priority: priority, - Reason: reason, - Size: size, - RequestedAt: time.Now(), - Source: "predictive", - }: - // Successfully queued - default: - // Queue full, skip warming - } -} - -// prefetchWorker processes prefetch requests -func (pcm *PredictiveCacheManager) prefetchWorker() { - defer pcm.wg.Done() - - for { - select { - case <-pcm.ctx.Done(): - return - case req := <-pcm.prefetchQueue: - // Process prefetch request - pcm.processPrefetchRequest(req) - } - } -} - -// analysisWorker performs periodic analysis and cache warming -func (pcm *PredictiveCacheManager) analysisWorker() { - defer pcm.wg.Done() - - ticker := time.NewTicker(30 * time.Second) // Analyze every 30 seconds - defer ticker.Stop() - - for { - select { - case <-pcm.ctx.Done(): - return - case <-ticker.C: - pcm.performAnalysis() - } - } -} - -// processPrefetchRequest processes a prefetch request -func (pcm *PredictiveCacheManager) processPrefetchRequest(req PrefetchRequest) { - // In a real implementation, this would: - // 1. Check if content is already cached - // 2. If not, fetch and cache it - // 3. Update statistics - - // For now, just log the prefetch request - // In production, integrate with the actual cache system -} - -// performAnalysis performs periodic analysis and cache warming -func (pcm *PredictiveCacheManager) performAnalysis() { - // Get popular content for warming - popular := pcm.cacheWarmer.GetPopularContent(10) - - // Request warming for popular content - for _, content := range popular { - if content.AccessCount > 5 { // Only warm frequently accessed content - select { - case pcm.cacheWarmer.warmerQueue <- WarmRequest{ - Key: content.Key, - Priority: content.Priority, - Reason: "popular_content", - }: - default: - // Queue full, skip - } - } - } -} - -// GetStats returns predictive caching statistics -func (pcm *PredictiveCacheManager) GetStats() *PredictiveStats { - pcm.stats.mu.RLock() - defer pcm.stats.mu.RUnlock() - - return &PredictiveStats{ - PrefetchHits: atomic.LoadInt64(&pcm.stats.PrefetchHits), - PrefetchMisses: atomic.LoadInt64(&pcm.stats.PrefetchMisses), - PrefetchRequests: atomic.LoadInt64(&pcm.stats.PrefetchRequests), - CacheWarmHits: atomic.LoadInt64(&pcm.stats.CacheWarmHits), - CacheWarmMisses: atomic.LoadInt64(&pcm.stats.CacheWarmMisses), - } -} - -// Stop stops the predictive cache manager -func (pcm *PredictiveCacheManager) Stop() { - pcm.cancel() - pcm.wg.Wait() -} diff --git a/vfs/predictive/predictive_test.go b/vfs/predictive/predictive_test.go deleted file mode 100644 index 8455611..0000000 --- a/vfs/predictive/predictive_test.go +++ /dev/null @@ -1,41 +0,0 @@ -package predictive - -import ( - "testing" -) - -func TestAccessPredictor_Basic(t *testing.T) { - t.Parallel() - p := NewAccessPredictor() - p.RecordSequence("a/b/c1", "a/b/c2") - next := p.PredictNext("a/b/c1") - if len(next) == 0 { - t.Log("no predictions (cold start ok)") - } - _ = p.IsPredictedAccess("a/b/c2") -} - -func TestCacheWarmer_Basic(t *testing.T) { - t.Parallel() - cw := NewCacheWarmer() - cw.RecordAccess("k1", 100) - cw.RecordAccess("k1", 100) - pop := cw.GetPopularContent(5) - _ = len(pop) - _ = NewWarmingStats() - _ = NewActiveWarmer("k", 1, "test") -} - -// TestPredictiveCacheManager_ConstructAndStop exercises New + RecordAccess under load + worker + Stop (no leak/panic; issue11). -func TestPredictiveCacheManager_ConstructAndStop(t *testing.T) { - t.Parallel() - pm := NewPredictiveCacheManager() - for i := 0; i < 20; i++ { - k := "k" + string(rune('0'+i%5)) - pm.RecordAccess(k, "", 100) // use actual API (RecordAccess); exercises warmer+predictor paths - } - // Stop exercises wg + cancel for workers - pm.Stop() - // double stop safe - pm.Stop() -}