mirror of
https://github.com/truewhile/MeBox.git
synced 2026-09-29 03:26:37 +08:00
Add configurable memory limits and LRU eviction for runtime cache
This commit is contained in:
@@ -46,6 +46,9 @@ func TestLoadDefaults(t *testing.T) {
|
||||
if cfg.Cache.MediaTTLSeconds != 90 {
|
||||
t.Fatalf("expected default media cache ttl 90, got %d", cfg.Cache.MediaTTLSeconds)
|
||||
}
|
||||
if cfg.Cache.MemoryMaxSizeMB != DefaultCacheMemoryMaxSizeMB {
|
||||
t.Fatalf("expected default runtime cache memory limit %d, got %d", DefaultCacheMemoryMaxSizeMB, cfg.Cache.MemoryMaxSizeMB)
|
||||
}
|
||||
if cfg.Search.Index != "mebox_media" {
|
||||
t.Fatalf("expected default search index, got %q", cfg.Search.Index)
|
||||
}
|
||||
@@ -88,6 +91,7 @@ func TestEnvOverride(t *testing.T) {
|
||||
t.Setenv("MEBOX_DATABASE_DSN", "postgres://mebox:secret@postgres:5432/mebox?sslmode=disable")
|
||||
t.Setenv("MEBOX_CACHE_REDIS_URL", "redis://redis:6379/0")
|
||||
t.Setenv("MEBOX_CACHE_MEDIA_TTL_SECONDS", "30")
|
||||
t.Setenv("MEBOX_CACHE_MEMORY_MAX_SIZE_MB", "48")
|
||||
t.Setenv("MEBOX_SEARCH_BACKEND", "opensearch")
|
||||
t.Setenv("MEBOX_SEARCH_OPENSEARCH_URL", "http://opensearch:9200")
|
||||
t.Setenv("MEBOX_LICENSE_SERVER_URL", "https://license.example.com")
|
||||
@@ -103,8 +107,8 @@ func TestEnvOverride(t *testing.T) {
|
||||
if cfg.Database.Type != "postgres" || cfg.Database.DSN == "" {
|
||||
t.Fatalf("expected postgres database config from env, got type=%q dsn=%q", cfg.Database.Type, cfg.Database.DSN)
|
||||
}
|
||||
if cfg.Cache.RedisURL != "redis://redis:6379/0" || cfg.Cache.MediaTTLSeconds != 30 {
|
||||
t.Fatalf("expected redis cache config from env, got url=%q ttl=%d", cfg.Cache.RedisURL, cfg.Cache.MediaTTLSeconds)
|
||||
if cfg.Cache.RedisURL != "redis://redis:6379/0" || cfg.Cache.MediaTTLSeconds != 30 || cfg.Cache.MemoryMaxSizeMB != 48 {
|
||||
t.Fatalf("expected redis cache config from env, got url=%q ttl=%d memory=%d", cfg.Cache.RedisURL, cfg.Cache.MediaTTLSeconds, cfg.Cache.MemoryMaxSizeMB)
|
||||
}
|
||||
if cfg.Search.Backend != "opensearch" || cfg.Search.OpenSearchURL != "http://opensearch:9200" {
|
||||
t.Fatalf("expected opensearch config from env, got backend=%q url=%q", cfg.Search.Backend, cfg.Search.OpenSearchURL)
|
||||
|
||||
@@ -7,6 +7,7 @@ const (
|
||||
defaultDatabaseMaxIdleConns = 4
|
||||
defaultLicenseServerURL = "https://mgosever.3jzs.com"
|
||||
defaultLicensePublicKey = "MCowBQYDK2VwAyEABRXnXy+urjrbKit6Yu/HiezWgP0NdsZW3tsegJWRrtI="
|
||||
DefaultCacheMemoryMaxSizeMB = 128
|
||||
)
|
||||
|
||||
func setDefaults(v *viper.Viper) {
|
||||
@@ -44,6 +45,7 @@ func setDefaults(v *viper.Viper) {
|
||||
|
||||
v.SetDefault("cache.cache_dir", "./cache")
|
||||
v.SetDefault("cache.images_max_size_mb", 500)
|
||||
v.SetDefault("cache.memory_max_size_mb", DefaultCacheMemoryMaxSizeMB)
|
||||
v.SetDefault("cache.cleanup_interval_min", 60)
|
||||
v.SetDefault("cache.redis_url", "")
|
||||
v.SetDefault("cache.redis_prefix", "mebox")
|
||||
|
||||
@@ -47,6 +47,9 @@ func (c *Config) normalize() error {
|
||||
if c.Cache.ImagesMaxSizeMB < 0 {
|
||||
c.Cache.ImagesMaxSizeMB = 0
|
||||
}
|
||||
if c.Cache.MemoryMaxSizeMB <= 0 {
|
||||
c.Cache.MemoryMaxSizeMB = DefaultCacheMemoryMaxSizeMB
|
||||
}
|
||||
if c.Cache.RedisPrefix == "" {
|
||||
c.Cache.RedisPrefix = "mebox"
|
||||
}
|
||||
|
||||
@@ -113,10 +113,12 @@ type LoggingConfig struct {
|
||||
MaxBackups int `mapstructure:"max_backups"`
|
||||
}
|
||||
|
||||
// CacheConfig 控制磁盘转码/刮削缓存。
|
||||
// CacheConfig 控制磁盘转写缓存和进程内热缓存。
|
||||
type CacheConfig struct {
|
||||
CacheDir string `mapstructure:"cache_dir"`
|
||||
ImagesMaxSizeMB int `mapstructure:"images_max_size_mb"`
|
||||
CacheDir string `mapstructure:"cache_dir"`
|
||||
ImagesMaxSizeMB int `mapstructure:"images_max_size_mb"`
|
||||
// MemoryMaxSizeMB 限制进程内 L1 缓存总字节数,JSON/对象缓存共用该预算。
|
||||
MemoryMaxSizeMB int `mapstructure:"memory_max_size_mb"`
|
||||
MaxDiskUsageMB int `mapstructure:"max_disk_usage_mb"`
|
||||
TTLHours int `mapstructure:"ttl_hours"`
|
||||
AutoCleanup bool `mapstructure:"auto_cleanup"`
|
||||
|
||||
@@ -68,6 +68,9 @@ func updateSettingHandler(svc *service.Container) gin.HandlerFunc {
|
||||
if req.Key == "cache.images_max_size_mb" && svc.Scheduler != nil {
|
||||
_ = svc.Scheduler.RunNowAsync(c.Request.Context(), "image_cache_cleanup")
|
||||
}
|
||||
if req.Key == "cache.memory_max_size_mb" && svc.Cache != nil {
|
||||
svc.Cache.SetMaxSizeMB(svc.Cfg.Cache.MemoryMaxSizeMB)
|
||||
}
|
||||
c.Status(http.StatusNoContent)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package service
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -13,20 +14,32 @@ import (
|
||||
"github.com/truewhile/MeBox/internal/config"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultRuntimeCacheEntryLimit = 2048
|
||||
runtimeCacheEntryOverheadBytes = 96
|
||||
runtimeCacheMapEntryOverheadBytes = 64
|
||||
runtimeCacheStringHeaderBytes = 16
|
||||
runtimeCacheEstimateMaxDepth = 64
|
||||
)
|
||||
|
||||
type RuntimeCacheService struct {
|
||||
log *zap.Logger
|
||||
client *redis.Client
|
||||
prefix string
|
||||
|
||||
mu sync.RWMutex
|
||||
memory map[string]runtimeCacheItem
|
||||
obj map[string]runtimeObjectItem
|
||||
limit int
|
||||
mu sync.RWMutex
|
||||
memory map[string]runtimeCacheItem
|
||||
obj map[string]runtimeObjectItem
|
||||
limit int
|
||||
maxBytes int64
|
||||
bytesUsed int64
|
||||
}
|
||||
|
||||
type runtimeCacheItem struct {
|
||||
raw []byte
|
||||
expiresAt time.Time
|
||||
lastUsed time.Time
|
||||
size int64
|
||||
}
|
||||
|
||||
// runtimeObjectItem 直存 Go 对象,跳过 JSON 编解码。热点路径(整库行、
|
||||
@@ -37,10 +50,22 @@ type runtimeCacheItem struct {
|
||||
type runtimeObjectItem struct {
|
||||
value any
|
||||
expiresAt time.Time
|
||||
lastUsed time.Time
|
||||
size int64
|
||||
}
|
||||
|
||||
func NewRuntimeCacheService(cfg *config.Config, log *zap.Logger) *RuntimeCacheService {
|
||||
c := &RuntimeCacheService{log: log, memory: map[string]runtimeCacheItem{}, obj: map[string]runtimeObjectItem{}, limit: 2048}
|
||||
maxBytes := int64(config.DefaultCacheMemoryMaxSizeMB) * 1024 * 1024
|
||||
if cfg != nil && cfg.Cache.MemoryMaxSizeMB > 0 {
|
||||
maxBytes = int64(cfg.Cache.MemoryMaxSizeMB) * 1024 * 1024
|
||||
}
|
||||
c := &RuntimeCacheService{
|
||||
log: log,
|
||||
memory: map[string]runtimeCacheItem{},
|
||||
obj: map[string]runtimeObjectItem{},
|
||||
limit: defaultRuntimeCacheEntryLimit,
|
||||
maxBytes: maxBytes,
|
||||
}
|
||||
if cfg == nil {
|
||||
return c
|
||||
}
|
||||
@@ -101,7 +126,7 @@ func (c *RuntimeCacheService) GetJSON(ctx context.Context, key string, out any)
|
||||
if json.Unmarshal(raw, out) != nil {
|
||||
return false
|
||||
}
|
||||
c.setMemory(fullKey, raw, 2*time.Second)
|
||||
c.setMemoryOwned(fullKey, raw, 2*time.Second)
|
||||
return true
|
||||
}
|
||||
}
|
||||
@@ -117,7 +142,7 @@ func (c *RuntimeCacheService) SetJSON(ctx context.Context, key string, value any
|
||||
return
|
||||
}
|
||||
fullKey := c.key(key)
|
||||
c.setMemory(fullKey, raw, ttl)
|
||||
c.setMemoryOwned(fullKey, raw, ttl)
|
||||
if c.client != nil {
|
||||
_ = c.client.Set(ctx, fullKey, raw, ttl).Err()
|
||||
}
|
||||
@@ -130,41 +155,70 @@ func (c *RuntimeCacheService) GetObject(key string) (any, bool) {
|
||||
}
|
||||
fullKey := c.key(key)
|
||||
now := time.Now()
|
||||
c.mu.RLock()
|
||||
c.mu.Lock()
|
||||
item, ok := c.obj[fullKey]
|
||||
c.mu.RUnlock()
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
if now.After(item.expiresAt) {
|
||||
c.mu.Lock()
|
||||
delete(c.obj, fullKey)
|
||||
c.mu.Unlock()
|
||||
return nil, false
|
||||
}
|
||||
if !now.Before(item.expiresAt) {
|
||||
c.removeObjectLocked(fullKey)
|
||||
c.mu.Unlock()
|
||||
return nil, false
|
||||
}
|
||||
item.lastUsed = now
|
||||
c.obj[fullKey] = item
|
||||
c.mu.Unlock()
|
||||
return item.value, true
|
||||
}
|
||||
|
||||
// SetObject 存入一个此后视为不可变的对象。
|
||||
// SetObject 存入一个此后视为不可变的对象。对象会按估算内存计入进程内缓存
|
||||
// 总预算;单个对象超过预算时不会进入缓存,避免一次大列表请求再次打爆内存。
|
||||
func (c *RuntimeCacheService) SetObject(key string, value any, ttl time.Duration) {
|
||||
if !c.Enabled() || strings.TrimSpace(key) == "" || value == nil || ttl <= 0 {
|
||||
return
|
||||
}
|
||||
fullKey := c.key(key)
|
||||
size := estimateRuntimeCacheObjectSize(fullKey, value)
|
||||
now := time.Now()
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if len(c.obj) >= c.limit {
|
||||
for k, item := range c.obj {
|
||||
if now.After(item.expiresAt) || len(c.obj) >= c.limit {
|
||||
delete(c.obj, k)
|
||||
}
|
||||
if len(c.obj) < c.limit {
|
||||
break
|
||||
}
|
||||
if size <= 0 || size > c.maxBytes {
|
||||
return
|
||||
}
|
||||
c.removeObjectLocked(fullKey)
|
||||
if !c.makeRoomLocked(now, size) {
|
||||
return
|
||||
}
|
||||
c.obj[fullKey] = runtimeObjectItem{
|
||||
value: value,
|
||||
expiresAt: now.Add(ttl),
|
||||
lastUsed: now,
|
||||
size: size,
|
||||
}
|
||||
c.bytesUsed += size
|
||||
}
|
||||
|
||||
// SetMaxSizeMB 热更新进程内缓存总预算。降低上限时会立即淘汰最久未使用的
|
||||
// 条目;传 0 或负数时恢复默认预算。
|
||||
func (c *RuntimeCacheService) SetMaxSizeMB(maxMB int) {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
if maxMB <= 0 {
|
||||
maxMB = config.DefaultCacheMemoryMaxSizeMB
|
||||
}
|
||||
maxBytes := int64(maxMB) * 1024 * 1024
|
||||
now := time.Now()
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.maxBytes = maxBytes
|
||||
c.evictExpiredLocked(now)
|
||||
for c.bytesUsed > c.maxBytes {
|
||||
if !c.evictOldestLocked() {
|
||||
break
|
||||
}
|
||||
}
|
||||
c.obj[fullKey] = runtimeObjectItem{value: value, expiresAt: now.Add(ttl)}
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) DeletePrefix(ctx context.Context, prefix string) {
|
||||
@@ -202,39 +256,145 @@ func (c *RuntimeCacheService) key(key string) string {
|
||||
|
||||
func (c *RuntimeCacheService) getMemory(key string) ([]byte, bool) {
|
||||
now := time.Now()
|
||||
c.mu.RLock()
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
item, ok := c.memory[key]
|
||||
c.mu.RUnlock()
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
if now.After(item.expiresAt) {
|
||||
c.mu.Lock()
|
||||
delete(c.memory, key)
|
||||
c.mu.Unlock()
|
||||
if !now.Before(item.expiresAt) {
|
||||
c.removeMemoryLocked(key)
|
||||
return nil, false
|
||||
}
|
||||
item.lastUsed = now
|
||||
c.memory[key] = item
|
||||
return item.raw, true
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) setMemory(key string, raw []byte, ttl time.Duration) {
|
||||
c.setMemoryBytes(key, raw, ttl, false)
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) setMemoryOwned(key string, raw []byte, ttl time.Duration) {
|
||||
c.setMemoryBytes(key, raw, ttl, true)
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) setMemoryBytes(key string, raw []byte, ttl time.Duration, owned bool) {
|
||||
if ttl <= 0 || len(raw) == 0 {
|
||||
return
|
||||
}
|
||||
size := int64(len(key)+len(raw)) + runtimeCacheEntryOverheadBytes
|
||||
now := time.Now()
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if len(c.memory) >= c.limit {
|
||||
now := time.Now()
|
||||
for k, item := range c.memory {
|
||||
if now.After(item.expiresAt) || len(c.memory) >= c.limit {
|
||||
delete(c.memory, k)
|
||||
}
|
||||
if len(c.memory) < c.limit {
|
||||
break
|
||||
}
|
||||
if size <= 0 || size > c.maxBytes {
|
||||
return
|
||||
}
|
||||
c.removeMemoryLocked(key)
|
||||
if !c.makeRoomLocked(now, size) {
|
||||
return
|
||||
}
|
||||
if !owned {
|
||||
raw = append([]byte(nil), raw...)
|
||||
}
|
||||
c.memory[key] = runtimeCacheItem{
|
||||
raw: raw,
|
||||
expiresAt: now.Add(ttl),
|
||||
lastUsed: now,
|
||||
size: size,
|
||||
}
|
||||
c.bytesUsed += size
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) makeRoomLocked(now time.Time, incoming int64) bool {
|
||||
if incoming <= 0 {
|
||||
incoming = 1
|
||||
}
|
||||
if c.limit <= 0 {
|
||||
c.limit = defaultRuntimeCacheEntryLimit
|
||||
}
|
||||
if c.maxBytes <= 0 {
|
||||
c.maxBytes = int64(config.DefaultCacheMemoryMaxSizeMB) * 1024 * 1024
|
||||
}
|
||||
if incoming > c.maxBytes {
|
||||
return false
|
||||
}
|
||||
c.evictExpiredLocked(now)
|
||||
for c.entryCountLocked() >= c.limit || c.bytesUsed+incoming > c.maxBytes {
|
||||
if !c.evictOldestLocked() {
|
||||
return false
|
||||
}
|
||||
}
|
||||
c.memory[key] = runtimeCacheItem{raw: append([]byte(nil), raw...), expiresAt: time.Now().Add(ttl)}
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) entryCountLocked() int {
|
||||
return len(c.memory) + len(c.obj)
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) evictExpiredLocked(now time.Time) {
|
||||
for key, item := range c.memory {
|
||||
if !now.Before(item.expiresAt) {
|
||||
c.removeMemoryLocked(key)
|
||||
}
|
||||
}
|
||||
for key, item := range c.obj {
|
||||
if !now.Before(item.expiresAt) {
|
||||
c.removeObjectLocked(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) evictOldestLocked() bool {
|
||||
var (
|
||||
oldestKey string
|
||||
oldestKind byte
|
||||
oldestAt time.Time
|
||||
)
|
||||
for key, item := range c.memory {
|
||||
if oldestKind == 0 || item.lastUsed.Before(oldestAt) {
|
||||
oldestKey, oldestKind, oldestAt = key, 'm', item.lastUsed
|
||||
}
|
||||
}
|
||||
for key, item := range c.obj {
|
||||
if oldestKind == 0 || item.lastUsed.Before(oldestAt) {
|
||||
oldestKey, oldestKind, oldestAt = key, 'o', item.lastUsed
|
||||
}
|
||||
}
|
||||
switch oldestKind {
|
||||
case 'm':
|
||||
c.removeMemoryLocked(oldestKey)
|
||||
return true
|
||||
case 'o':
|
||||
c.removeObjectLocked(oldestKey)
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) removeMemoryLocked(key string) {
|
||||
item, ok := c.memory[key]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
delete(c.memory, key)
|
||||
c.bytesUsed -= item.size
|
||||
if c.bytesUsed < 0 {
|
||||
c.bytesUsed = 0
|
||||
}
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) removeObjectLocked(key string) {
|
||||
item, ok := c.obj[key]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
delete(c.obj, key)
|
||||
c.bytesUsed -= item.size
|
||||
if c.bytesUsed < 0 {
|
||||
c.bytesUsed = 0
|
||||
}
|
||||
}
|
||||
|
||||
func (c *RuntimeCacheService) deleteMemoryPrefix(prefix string) {
|
||||
@@ -242,12 +402,137 @@ func (c *RuntimeCacheService) deleteMemoryPrefix(prefix string) {
|
||||
defer c.mu.Unlock()
|
||||
for key := range c.memory {
|
||||
if strings.HasPrefix(key, prefix) {
|
||||
delete(c.memory, key)
|
||||
c.removeMemoryLocked(key)
|
||||
}
|
||||
}
|
||||
for key := range c.obj {
|
||||
if strings.HasPrefix(key, prefix) {
|
||||
delete(c.obj, key)
|
||||
c.removeObjectLocked(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type runtimeCacheVisit struct {
|
||||
kind reflect.Kind
|
||||
ptr uintptr
|
||||
}
|
||||
|
||||
func estimateRuntimeCacheObjectSize(key string, value any) int64 {
|
||||
seen := make(map[runtimeCacheVisit]struct{})
|
||||
size := int64(len(key)) + runtimeCacheEntryOverheadBytes
|
||||
size += estimateRuntimeCacheReflectSize(reflect.ValueOf(value), seen, 0)
|
||||
return size
|
||||
}
|
||||
|
||||
func estimateRuntimeCacheReflectSize(v reflect.Value, seen map[runtimeCacheVisit]struct{}, depth int) int64 {
|
||||
if !v.IsValid() || depth > runtimeCacheEstimateMaxDepth {
|
||||
return 0
|
||||
}
|
||||
switch v.Kind() {
|
||||
case reflect.Interface:
|
||||
if v.IsNil() {
|
||||
return 0
|
||||
}
|
||||
return estimateRuntimeCacheReflectSize(v.Elem(), seen, depth+1)
|
||||
case reflect.Pointer:
|
||||
if v.IsNil() || markRuntimeCacheVisit(v, seen) {
|
||||
return 0
|
||||
}
|
||||
return int64(v.Type().Elem().Size()) + estimateRuntimeCacheNestedSize(v.Elem(), seen, depth+1)
|
||||
case reflect.String:
|
||||
return int64(len(v.String())) + runtimeCacheStringHeaderBytes
|
||||
case reflect.Slice:
|
||||
if v.IsNil() || markRuntimeCacheVisit(v, seen) {
|
||||
return 0
|
||||
}
|
||||
total := int64(v.Type().Size()) + int64(v.Cap())*int64(v.Type().Elem().Size())
|
||||
for i := 0; i < v.Len(); i++ {
|
||||
total += estimateRuntimeCacheNestedSize(v.Index(i), seen, depth+1)
|
||||
}
|
||||
return total
|
||||
case reflect.Map:
|
||||
if v.IsNil() || markRuntimeCacheVisit(v, seen) {
|
||||
return 0
|
||||
}
|
||||
total := int64(v.Type().Size()) + int64(v.Len())*runtimeCacheMapEntryOverheadBytes
|
||||
iter := v.MapRange()
|
||||
for iter.Next() {
|
||||
total += estimateRuntimeCacheNestedSize(iter.Key(), seen, depth+1)
|
||||
total += estimateRuntimeCacheNestedSize(iter.Value(), seen, depth+1)
|
||||
}
|
||||
return total
|
||||
case reflect.Struct:
|
||||
total := int64(v.Type().Size())
|
||||
for i := 0; i < v.NumField(); i++ {
|
||||
total += estimateRuntimeCacheNestedSize(v.Field(i), seen, depth+1)
|
||||
}
|
||||
return total
|
||||
case reflect.Array:
|
||||
total := int64(v.Type().Size())
|
||||
for i := 0; i < v.Len(); i++ {
|
||||
total += estimateRuntimeCacheNestedSize(v.Index(i), seen, depth+1)
|
||||
}
|
||||
return total
|
||||
default:
|
||||
return int64(v.Type().Size())
|
||||
}
|
||||
}
|
||||
|
||||
func estimateRuntimeCacheNestedSize(v reflect.Value, seen map[runtimeCacheVisit]struct{}, depth int) int64 {
|
||||
if !v.IsValid() || depth > runtimeCacheEstimateMaxDepth {
|
||||
return 0
|
||||
}
|
||||
switch v.Kind() {
|
||||
case reflect.Interface, reflect.Pointer:
|
||||
return estimateRuntimeCacheReflectSize(v, seen, depth+1)
|
||||
case reflect.String:
|
||||
return int64(len(v.String()))
|
||||
case reflect.Slice:
|
||||
if v.IsNil() || markRuntimeCacheVisit(v, seen) {
|
||||
return 0
|
||||
}
|
||||
total := int64(v.Cap()) * int64(v.Type().Elem().Size())
|
||||
for i := 0; i < v.Len(); i++ {
|
||||
total += estimateRuntimeCacheNestedSize(v.Index(i), seen, depth+1)
|
||||
}
|
||||
return total
|
||||
case reflect.Map:
|
||||
if v.IsNil() || markRuntimeCacheVisit(v, seen) {
|
||||
return 0
|
||||
}
|
||||
total := int64(v.Len()) * runtimeCacheMapEntryOverheadBytes
|
||||
iter := v.MapRange()
|
||||
for iter.Next() {
|
||||
total += estimateRuntimeCacheNestedSize(iter.Key(), seen, depth+1)
|
||||
total += estimateRuntimeCacheNestedSize(iter.Value(), seen, depth+1)
|
||||
}
|
||||
return total
|
||||
case reflect.Struct:
|
||||
total := int64(0)
|
||||
for i := 0; i < v.NumField(); i++ {
|
||||
total += estimateRuntimeCacheNestedSize(v.Field(i), seen, depth+1)
|
||||
}
|
||||
return total
|
||||
case reflect.Array:
|
||||
total := int64(0)
|
||||
for i := 0; i < v.Len(); i++ {
|
||||
total += estimateRuntimeCacheNestedSize(v.Index(i), seen, depth+1)
|
||||
}
|
||||
return total
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
func markRuntimeCacheVisit(v reflect.Value, seen map[runtimeCacheVisit]struct{}) bool {
|
||||
ptr := v.Pointer()
|
||||
if ptr == 0 {
|
||||
return false
|
||||
}
|
||||
visit := runtimeCacheVisit{kind: v.Kind(), ptr: ptr}
|
||||
if _, ok := seen[visit]; ok {
|
||||
return true
|
||||
}
|
||||
seen[visit] = struct{}{}
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/truewhile/MeBox/internal/config"
|
||||
"github.com/truewhile/MeBox/internal/model"
|
||||
)
|
||||
|
||||
func newRuntimeCacheForTest(t *testing.T, maxMB int) *RuntimeCacheService {
|
||||
t.Helper()
|
||||
cfg := &config.Config{}
|
||||
cfg.Cache.MemoryMaxSizeMB = maxMB
|
||||
return NewRuntimeCacheService(cfg, zap.NewNop())
|
||||
}
|
||||
|
||||
func TestRuntimeCacheEvictsByByteBudget(t *testing.T) {
|
||||
cache := newRuntimeCacheForTest(t, 1)
|
||||
cache.SetObject("large", strings.Repeat("a", 800<<10), time.Minute)
|
||||
cache.SetObject("small", strings.Repeat("b", 300<<10), time.Minute)
|
||||
|
||||
if cache.bytesUsed > cache.maxBytes {
|
||||
t.Fatalf("cache bytes=%d exceeds max=%d", cache.bytesUsed, cache.maxBytes)
|
||||
}
|
||||
if _, ok := cache.GetObject("large"); ok {
|
||||
t.Fatal("expected oldest large entry to be evicted")
|
||||
}
|
||||
if _, ok := cache.GetObject("small"); !ok {
|
||||
t.Fatal("expected newest small entry to remain cached")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRuntimeCacheSkipsOversizedObject(t *testing.T) {
|
||||
cache := newRuntimeCacheForTest(t, 1)
|
||||
cache.SetObject("huge", strings.Repeat("x", 2<<20), time.Minute)
|
||||
|
||||
if _, ok := cache.GetObject("huge"); ok {
|
||||
t.Fatal("oversized object must not be cached")
|
||||
}
|
||||
if cache.bytesUsed != 0 {
|
||||
t.Fatalf("bytesUsed=%d, want 0", cache.bytesUsed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRuntimeCachePrefixDeleteReleasesByteBudget(t *testing.T) {
|
||||
cache := newRuntimeCacheForTest(t, 2)
|
||||
for i := 0; i < 4; i++ {
|
||||
cache.SetJSON(context.Background(), fmt.Sprintf("media:%d", i), strings.Repeat("x", 64<<10), time.Minute)
|
||||
}
|
||||
if cache.bytesUsed == 0 {
|
||||
t.Fatal("expected cached bytes")
|
||||
}
|
||||
cache.DeletePrefix(context.Background(), "media:")
|
||||
if cache.bytesUsed != 0 {
|
||||
t.Fatalf("bytesUsed=%d after prefix delete, want 0", cache.bytesUsed)
|
||||
}
|
||||
for i := 0; i < 4; i++ {
|
||||
var out string
|
||||
if cache.GetJSON(context.Background(), fmt.Sprintf("media:%d", i), &out) {
|
||||
t.Fatalf("entry media:%d should have been deleted", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRuntimeCacheStillEnforcesEntryLimit(t *testing.T) {
|
||||
cache := newRuntimeCacheForTest(t, 1)
|
||||
cache.limit = 3
|
||||
for i := 0; i < 5; i++ {
|
||||
cache.SetObject(fmt.Sprintf("item-%d", i), i, time.Minute)
|
||||
}
|
||||
if got := cache.entryCountLocked(); got != 3 {
|
||||
t.Fatalf("entry count=%d, want 3", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRuntimeCacheEstimatesMediaRows(t *testing.T) {
|
||||
rows := []model.Media{{
|
||||
Title: strings.Repeat("t", 1024),
|
||||
Path: "/media/movies/example.mkv",
|
||||
PosterURL: "https://image.example/poster.jpg",
|
||||
BackdropURL: "https://image.example/backdrop.jpg",
|
||||
Overview: strings.Repeat("o", 2048),
|
||||
Genres: "Action,Adventure",
|
||||
}}
|
||||
size := estimateRuntimeCacheObjectSize("media:test", rows)
|
||||
if size < int64(len(rows[0].Title)+len(rows[0].Overview)) {
|
||||
t.Fatalf("estimated size %d is smaller than payload", size)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRuntimeCacheSetMaxSizeEvictsImmediately(t *testing.T) {
|
||||
cache := newRuntimeCacheForTest(t, 2)
|
||||
cache.SetObject("old", strings.Repeat("a", 700<<10), time.Minute)
|
||||
cache.SetObject("new", strings.Repeat("b", 700<<10), time.Minute)
|
||||
if cache.bytesUsed <= 1<<20 {
|
||||
t.Fatalf("test setup bytesUsed=%d, want >1MiB", cache.bytesUsed)
|
||||
}
|
||||
|
||||
cache.SetMaxSizeMB(1)
|
||||
if cache.maxBytes != 1<<20 {
|
||||
t.Fatalf("maxBytes=%d, want 1MiB", cache.maxBytes)
|
||||
}
|
||||
if cache.bytesUsed > cache.maxBytes {
|
||||
t.Fatalf("bytesUsed=%d exceeds maxBytes=%d", cache.bytesUsed, cache.maxBytes)
|
||||
}
|
||||
if _, ok := cache.GetObject("old"); ok {
|
||||
t.Fatal("oldest entry should be evicted after lowering cache limit")
|
||||
}
|
||||
if _, ok := cache.GetObject("new"); !ok {
|
||||
t.Fatal("newest entry should remain after lowering cache limit")
|
||||
}
|
||||
}
|
||||
@@ -119,6 +119,22 @@ func ApplyRuntimeSetting(cfg *config.Config, key, value string) {
|
||||
}
|
||||
cfg.Cache.ImagesMaxSizeMB = n
|
||||
}
|
||||
case "cache.memory_max_size_mb":
|
||||
n := config.DefaultCacheMemoryMaxSizeMB
|
||||
if value != "" {
|
||||
parsed, err := strconv.Atoi(value)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
n = parsed
|
||||
}
|
||||
if n <= 0 {
|
||||
n = config.DefaultCacheMemoryMaxSizeMB
|
||||
}
|
||||
if n > 4096 {
|
||||
n = 4096
|
||||
}
|
||||
cfg.Cache.MemoryMaxSizeMB = n
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -41,3 +41,22 @@ func TestApplyRuntimeSettingTranscodeSwitches(t *testing.T) {
|
||||
t.Fatalf("max cpu threads = %d, want clamp 1", cfg.App.MaxCPUThreads)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyRuntimeSettingMemoryCacheLimit(t *testing.T) {
|
||||
cfg := &config.Config{}
|
||||
|
||||
ApplyRuntimeSetting(cfg, "cache.memory_max_size_mb", "64")
|
||||
if cfg.Cache.MemoryMaxSizeMB != 64 {
|
||||
t.Fatalf("memory cache limit = %d, want 64", cfg.Cache.MemoryMaxSizeMB)
|
||||
}
|
||||
|
||||
ApplyRuntimeSetting(cfg, "cache.memory_max_size_mb", "")
|
||||
if cfg.Cache.MemoryMaxSizeMB != config.DefaultCacheMemoryMaxSizeMB {
|
||||
t.Fatalf("empty value should restore default, got %d", cfg.Cache.MemoryMaxSizeMB)
|
||||
}
|
||||
|
||||
ApplyRuntimeSetting(cfg, "cache.memory_max_size_mb", "99999")
|
||||
if cfg.Cache.MemoryMaxSizeMB != 4096 {
|
||||
t.Fatalf("memory cache limit should clamp to 4096, got %d", cfg.Cache.MemoryMaxSizeMB)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,53 +8,95 @@ import (
|
||||
"github.com/truewhile/MeBox/internal/model"
|
||||
)
|
||||
|
||||
// existingLocalMediaSnapshot 保留整库快照入口,主要供兼容性调用和测试使用。
|
||||
// 实际扫库路径使用 existingLocalMediaSnapshotForRoot,避免多根目录库重复持有
|
||||
// 与当前 root 无关的媒体记录。
|
||||
func (s *ScannerService) existingLocalMediaSnapshot(ctx context.Context, libraryID string) (map[string]existingLocalMedia, error) {
|
||||
var rows []model.Media
|
||||
if err := s.repo.DB.WithContext(ctx).
|
||||
return s.existingLocalMediaSnapshotForRoot(ctx, libraryID, "", "")
|
||||
}
|
||||
|
||||
// existingLocalMediaSnapshotForRoot 以流式方式读取当前 root 的已有媒体索引,
|
||||
// 不再先构造 []model.Media 再复制成 map。map 中的字符串只在最终索引中保留一份,
|
||||
// 可以显著降低大库扫描时的峰值内存。
|
||||
func (s *ScannerService) existingLocalMediaSnapshotForRoot(ctx context.Context, libraryID, rootID, rootPath string) (map[string]existingLocalMedia, error) {
|
||||
if s == nil || s.repo == nil || s.repo.DB == nil {
|
||||
return map[string]existingLocalMedia{}, nil
|
||||
}
|
||||
libraryID = strings.TrimSpace(libraryID)
|
||||
rootID = strings.TrimSpace(rootID)
|
||||
rootPath = strings.TrimSpace(rootPath)
|
||||
query := s.repo.DB.WithContext(ctx).
|
||||
Model(&model.Media{}).
|
||||
Where("library_id = ? AND path NOT LIKE ?", libraryID, "cloud://%")
|
||||
if rootID != "" {
|
||||
// 兼容尚未回填 library_root_id 的旧数据;没有 root id 的行再由路径归属过滤。
|
||||
query = query.Where("(library_root_id = ? OR library_root_id = '' OR library_root_id IS NULL)", rootID)
|
||||
}
|
||||
rows, err := query.
|
||||
Select("path", "library_root_id", "relative_path", "title", "original_name", "episode_title", "size_bytes", "duration_sec", "width", "height", "video_codec", "audio_codec", "container", "strm_url", "file_id", "poster_url", "backdrop_url", "overview", "year", "release_date", "rating", "tm_db_id", "bangumi_id", "douban_id", "thetvdb_id", "season_num", "episode_num", "genres", "countries", "languages", "nsfw", "scrape_status").
|
||||
Where("library_id = ? AND path NOT LIKE ?", libraryID, "cloud://%").
|
||||
Find(&rows).Error; err != nil {
|
||||
Rows()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
snapshot := make(map[string]existingLocalMedia, len(rows))
|
||||
for _, row := range rows {
|
||||
if strings.TrimSpace(row.Path) == "" {
|
||||
defer rows.Close()
|
||||
|
||||
snapshot := make(map[string]existingLocalMedia)
|
||||
for rows.Next() {
|
||||
var (
|
||||
path string
|
||||
row existingLocalMedia
|
||||
)
|
||||
if err := rows.Scan(
|
||||
&path,
|
||||
&row.LibraryRootID,
|
||||
&row.RelativePath,
|
||||
&row.Title,
|
||||
&row.OriginalName,
|
||||
&row.EpisodeTitle,
|
||||
&row.SizeBytes,
|
||||
&row.DurationSec,
|
||||
&row.Width,
|
||||
&row.Height,
|
||||
&row.VideoCodec,
|
||||
&row.AudioCodec,
|
||||
&row.Container,
|
||||
&row.STRMURL,
|
||||
&row.FileID,
|
||||
&row.PosterURL,
|
||||
&row.BackdropURL,
|
||||
&row.Overview,
|
||||
&row.Year,
|
||||
&row.ReleaseDate,
|
||||
&row.Rating,
|
||||
&row.TMDbID,
|
||||
&row.BangumiID,
|
||||
&row.DoubanID,
|
||||
&row.TheTVDBID,
|
||||
&row.SeasonNum,
|
||||
&row.EpisodeNum,
|
||||
&row.Genres,
|
||||
&row.Countries,
|
||||
&row.Languages,
|
||||
&row.NSFW,
|
||||
&row.ScrapeStatus,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
path = strings.TrimSpace(path)
|
||||
if path == "" {
|
||||
continue
|
||||
}
|
||||
snapshot[filepath.Clean(row.Path)] = existingLocalMedia{
|
||||
LibraryRootID: row.LibraryRootID,
|
||||
RelativePath: row.RelativePath,
|
||||
Title: row.Title,
|
||||
OriginalName: row.OriginalName,
|
||||
EpisodeTitle: row.EpisodeTitle,
|
||||
SizeBytes: row.SizeBytes,
|
||||
DurationSec: row.DurationSec,
|
||||
Width: row.Width,
|
||||
Height: row.Height,
|
||||
VideoCodec: row.VideoCodec,
|
||||
AudioCodec: row.AudioCodec,
|
||||
Container: row.Container,
|
||||
STRMURL: row.STRMURL,
|
||||
FileID: row.FileID,
|
||||
PosterURL: row.PosterURL,
|
||||
BackdropURL: row.BackdropURL,
|
||||
Overview: row.Overview,
|
||||
Year: row.Year,
|
||||
ReleaseDate: row.ReleaseDate,
|
||||
Rating: row.Rating,
|
||||
TMDbID: row.TMDbID,
|
||||
BangumiID: row.BangumiID,
|
||||
DoubanID: row.DoubanID,
|
||||
TheTVDBID: row.TheTVDBID,
|
||||
SeasonNum: row.SeasonNum,
|
||||
EpisodeNum: row.EpisodeNum,
|
||||
Genres: row.Genres,
|
||||
Countries: row.Countries,
|
||||
Languages: row.Languages,
|
||||
NSFW: row.NSFW,
|
||||
ScrapeStatus: row.ScrapeStatus,
|
||||
if rootPath != "" {
|
||||
if rootID == "" || strings.TrimSpace(row.LibraryRootID) == "" {
|
||||
if !pathBelongsToRoot(path, rootPath) {
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
snapshot[filepath.Clean(path)] = row
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return snapshot, nil
|
||||
}
|
||||
|
||||
@@ -59,3 +59,34 @@ func TestExistingLocalMediaSnapshotFiltersAndCleansLocalRows(t *testing.T) {
|
||||
t.Fatalf("identity fields not preserved: %#v", row)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExistingLocalMediaSnapshotForRootFiltersOtherRoots(t *testing.T) {
|
||||
db := newServiceTestDB(t, &model.Media{})
|
||||
repos := repository.New(db)
|
||||
scanner := NewScannerService(&config.Config{}, zap.NewNop(), repos, NewHub(zap.NewNop()), nil, nil)
|
||||
|
||||
root1 := t.TempDir()
|
||||
root2 := t.TempDir()
|
||||
path1 := filepath.Join(root1, "Root1.mkv")
|
||||
path2 := filepath.Join(root2, "Root2.mkv")
|
||||
if err := db.Create(&[]model.Media{
|
||||
{LibraryID: "lib-1", LibraryRootID: "root-1", Path: path1, SizeBytes: 11},
|
||||
{LibraryID: "lib-1", LibraryRootID: "root-2", Path: path2, SizeBytes: 22},
|
||||
}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
got, err := scanner.existingLocalMediaSnapshotForRoot(t.Context(), "lib-1", "root-1", root1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("snapshot len = %d, want 1: %#v", len(got), got)
|
||||
}
|
||||
if _, ok := got[filepath.Clean(path1)]; !ok {
|
||||
t.Fatalf("root-1 media missing from %#v", got)
|
||||
}
|
||||
if _, ok := got[filepath.Clean(path2)]; ok {
|
||||
t.Fatalf("root-2 media leaked into root-1 snapshot: %#v", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -73,11 +73,6 @@ func (s *ScannerService) scanLibrary(ctx context.Context, libraryID string, auto
|
||||
}
|
||||
res := &ScanResult{LibraryID: lib.ID}
|
||||
writeBatch := newLocalMediaWriteBatch(s, ctx, res, 100)
|
||||
existingMedia, err := s.existingLocalMediaSnapshot(ctx, lib.ID)
|
||||
if err != nil {
|
||||
s.log.Warn("load existing local media snapshot failed", zap.String("library_id", lib.ID), zap.Error(err))
|
||||
existingMedia = nil
|
||||
}
|
||||
|
||||
roots, err := s.localLibraryScanRoots(ctx, lib)
|
||||
if err != nil {
|
||||
@@ -109,6 +104,15 @@ func (s *ScannerService) scanLibrary(ctx context.Context, libraryID string, auto
|
||||
}
|
||||
continue
|
||||
}
|
||||
existingMedia, snapshotErr := s.existingLocalMediaSnapshotForRoot(ctx, lib.ID, root.ID, root.Path)
|
||||
if snapshotErr != nil {
|
||||
s.log.Warn("load existing local media snapshot failed",
|
||||
zap.String("library_id", lib.ID),
|
||||
zap.String("root_id", root.ID),
|
||||
zap.String("path", root.Path),
|
||||
zap.Error(snapshotErr))
|
||||
existingMedia = nil
|
||||
}
|
||||
seen, walkErr := s.scanLocalLibraryFiles(ctx, lib, &root, existingMedia, writeBatch, res)
|
||||
if walkErr != nil {
|
||||
addScanError(res, root.Path, walkErr)
|
||||
@@ -147,9 +151,13 @@ func (s *ScannerService) scanLocalLibraryRoot(ctx context.Context, lib *model.Li
|
||||
return res, err
|
||||
}
|
||||
writeBatch := newLocalMediaWriteBatch(s, ctx, res, 100)
|
||||
existingMedia, err := s.existingLocalMediaSnapshot(ctx, lib.ID)
|
||||
existingMedia, err := s.existingLocalMediaSnapshotForRoot(ctx, lib.ID, root.ID, root.Path)
|
||||
if err != nil {
|
||||
s.log.Warn("load existing local media snapshot failed", zap.String("library_id", lib.ID), zap.Error(err))
|
||||
s.log.Warn("load existing local media snapshot failed",
|
||||
zap.String("library_id", lib.ID),
|
||||
zap.String("root_id", root.ID),
|
||||
zap.String("path", root.Path),
|
||||
zap.Error(err))
|
||||
existingMedia = nil
|
||||
}
|
||||
seen, walkErr := s.scanLocalLibraryFiles(ctx, lib, root, existingMedia, writeBatch, res)
|
||||
|
||||
@@ -99,6 +99,14 @@ export const generalSettingsGroup: SettingGroup = {
|
||||
hint: 'NAS 建议 1;用于扫描、整理洗版和手动探测,避免同时启动多个 ffprobe 进程',
|
||||
defaultValue: '1',
|
||||
},
|
||||
{
|
||||
key: 'cache.memory_max_size_mb',
|
||||
label: '进程内缓存上限 (MB)',
|
||||
type: 'number',
|
||||
hint: 'JSON/列表等热缓存共用的内存上限。低配机器建议 32-64,内存充足可用 128-256;保存后会立即裁剪超额缓存,默认 128MB',
|
||||
defaultValue: '128',
|
||||
placeholder: '128',
|
||||
},
|
||||
{
|
||||
key: 'cache.images_max_size_mb',
|
||||
label: '图片缓存上限 (MB)',
|
||||
|
||||
Reference in New Issue
Block a user