chore(docsfilter): tombstone headers cache cleanup

Author: forshev

cmd/seq-db/seq-db.go

@@ -314,8 +314,9 @@ func startStore(
 			},
 		},
 		Filters: docsfilter.Config{
-			DataDir: cfg.DocsFilter.DataDir,
-			Workers: cfg.DocsFilter.Concurrency,
+			DataDir:        cfg.DocsFilter.DataDir,
+			Workers:        cfg.DocsFilter.Concurrency,
+			CacheSizeLimit: uint64(cfg.DocsFilter.CacheSize),
 		},
 	}
 

config/config.go

@@ -271,6 +271,7 @@ type Config struct {
 		DataDir     string   `config:"data_dir"`
 		Concurrency int      `config:"concurrency"`
 		Filters     []Filter `config:"filters"`
+		CacheSize   Bytes    `config:"cache_size" default:"100MiB"`
 	} `config:"docs_filter"`
 
 	// Experimental provides flags

docsfilter/docs_filter.go

@@ -30,15 +30,20 @@ const (
 	tmpExt         = ".tmp"
 )
 
-const defaultMaintenanceInterval = 30 * time.Second
+const (
+	defaultMaintenanceInterval = 30 * time.Second
+	defaultCacheCleanInterval  = 10 * time.Millisecond
+	defaultCacheGCDelay        = 1 * time.Second
+)
 
 type MappingProvider interface {
 	GetMapping() seq.Mapping
 }
 
 type Config struct {
-	DataDir string
-	Workers int
+	DataDir        string
+	Workers        int
+	CacheSizeLimit uint64
 }
 
 type DocsFilter struct {
@@ -56,8 +61,11 @@ type DocsFilter struct {
 	createDirOnce *sync.Once
 
 	maintenanceInterval time.Duration
+	cacheCleanInterval  time.Duration
+	cacheGCDelay        time.Duration
 
-	headersCache *cache.Cache[[]lidsBlockHeader]
+	headersCache        *cache.Cache[[]lidsBlockHeader]
+	headersCacheCleaner *cache.Cleaner
 }
 
 func New(
@@ -78,6 +86,8 @@ func New(
 		filtersMap[string(f.Hash())] = f
 	}
 
+	cacheCleaner := cache.NewCleaner(cfg.CacheSizeLimit, nil)
+
 	return &DocsFilter{
 		ctx:                 ctx,
 		config:              cfg,
@@ -88,8 +98,10 @@ func New(
 		rateLimit:           make(chan struct{}, workers),
 		createDirOnce:       &sync.Once{},
 		maintenanceInterval: defaultMaintenanceInterval,
-		// TODO: create cache properly (cleaner, metrics) (use cacheMaintainer ???)
-		headersCache: cache.NewCache[[]lidsBlockHeader](nil, nil),
+		cacheCleanInterval:  defaultCacheCleanInterval,
+		cacheGCDelay:        defaultCacheGCDelay,
+		headersCache:        cache.NewCache[[]lidsBlockHeader](cacheCleaner, nil),
+		headersCacheCleaner: cacheCleaner,
 	}
 }
 
@@ -107,6 +119,7 @@ func (df *DocsFilter) Start(fracs fracmanager.List) {
 	}
 
 	go df.maintenance()
+	go df.cacheCleanLoop()
 
 	mapping := df.mp.GetMapping()
 
@@ -382,6 +395,25 @@ func (df *DocsFilter) maintenance() {
 	}
 }
 
+func (df *DocsFilter) cacheCleanLoop() {
+	runs := 0
+	gcRunsCount := int(df.cacheGCDelay / df.cacheCleanInterval)
+
+	for {
+		runs++
+		df.headersCacheCleaner.Cleanup(&cache.CleanStat{})
+		df.headersCacheCleaner.Rotate()
+
+		if runs >= gcRunsCount {
+			runs = 0
+			df.headersCacheCleaner.CleanEmptyGenerations()
+			df.headersCacheCleaner.ReleaseBuckets()
+		}
+
+		time.Sleep(df.cacheCleanInterval)
+	}
+}
+
 func (df *DocsFilter) checkDiskUsage() {
 	du := int64(0)
 

frac/active.go

@@ -46,7 +46,7 @@ type Active struct {
 	TokenList *TokenList
 
 	DocsPositions *DocsPositions
-	IDsToLIDs     *ActiveLIDs // TODO: (???)
+	IDsToLIDs     *ActiveLIDs
 
 	docsFile   *os.File
 	docsReader storage.DocsReader