s1d3sw1ped/steamcache2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 20 Wiki Activity

Enhance DiskFS initialization and error handling

Browse Source 
- Updated `disk.New` to support asynchronous initialization for large caches, improving responsiveness during startup.
- Introduced an eviction function parameter to `disk.New`, ensuring proper handling of over-capacity scenarios.
- Enhanced error handling in various components, including memory and disk tests, to ensure robustness and clarity.
- Refactored tests to validate new behaviors, including checks for delayed attachment and proper error propagation.
- Removed obsolete error handling code and tests related to the now-deleted errors package, streamlining the codebase.
This commit is contained in:
Justin Harms
2026-05-27 13:15:33 -05:00
parent 4861f93e6f
commit feda55e225
18 changed files with 584 additions and 1380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+8
View File
@@ -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:
cmd/root.go
+5 -1
View File
@@ -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)
steamcache/errors/errors.go
-120
View File
@@ -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
}
steamcache/steamcache.go
+69 -51
View File
@@ -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())
}
steamcache/steamcache_test.go
+62 -21
View File
@@ -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()
}
}
vfs/adaptive/adaptive.go
-280
View File
@@ -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()
}
vfs/adaptive/adaptive_test.go
-47
View File
@@ -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()
}
vfs/cache/cache_test.go
Vendored
+24 -6
View File
@@ -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)
vfs/cachestate/cachestate.go
-5
View File
@@ -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
vfs/disk/disk.go
+141 -72
View File
@@ -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
vfs/disk/disk_test.go
+168 -9
View File
@@ -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)
}
}
vfs/eviction/eviction_test.go
+16 -4
View File
@@ -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()
vfs/gc/gc_test.go
+16 -4
View File
@@ -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()
vfs/memory/memory.go
+4 -3
View File
@@ -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
vfs/memory/memory_test.go
+71 -14
View File
@@ -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)
}
}
vfs/memory/monitor.go
-274
View File
@@ -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(),
}
}
vfs/predictive/predictive.go
-428
View File
@@ -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()
}
vfs/predictive/predictive_test.go
-41
View File
@@ -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()
}