feat(fracmanager): improve framanager consistency

Author: eguguchkin

cmd/distribution/main.go

@@ -161,7 +161,7 @@ func main() {
 		}
 	}
 
-	fc := fracmanager.NewSealedFracCache(filePathDist)
+	fc := fracmanager.NewFracInfoCache(filePathDist)
 
 	lastSavedTime := time.Now()
 	for _, path := range getAllFracs(dataDir) {
@@ -170,7 +170,7 @@ func main() {
 
 		logger.Info("start process", zap.String("name", key))
 
-		info, ok := fc.GetFracInfo(key)
+		info, ok := fc.Get(key)
 		if ok {
 			logger.Info("found in frac-cache", zap.String("key", key))
 		} else {
@@ -198,7 +198,7 @@ func main() {
 		}
 
 		buildDist(info.Distribution, path, info)
-		fc.AddFraction(key, info)
+		fc.Add(info)
 		logger.Info("built distribution", zap.Int("affected_minutes", len(info.Distribution.GetDist())))
 		printDistribution(info.Distribution)
 

cmd/index_analyzer/main.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"hash/fnv"
 	"os"
+	"sync"
 	"time"
 
 	"github.com/alecthomas/units"
@@ -57,7 +58,15 @@ func main() {
 func getCacheMaintainer() (*fracmanager.CacheMaintainer, func()) {
 	done := make(chan struct{})
 	cm := fracmanager.NewCacheMaintainer(uint64(units.GiB), uint64(units.MiB*64), nil)
-	wg := cm.RunCleanLoop(done, time.Second, time.Second)
+
+	wg := sync.WaitGroup{}
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		cm.RunCleanLoop(done, time.Second, time.Second)
+	}()
+
 	return cm, func() {
 		close(done)
 		wg.Wait()

cmd/seq-db/seq-db.go

@@ -251,7 +251,6 @@ func startStore(
 			CacheSize:         uint64(cfg.Resources.CacheSize),
 			SortCacheSize:     uint64(cfg.Resources.SortDocsCacheSize),
 			FracLoadLimit:     0,
-			ShouldReplay:      true,
 			MaintenanceDelay:  0,
 			CacheGCDelay:      0,
 			CacheCleanupDelay: 0,

consts/consts.go

@@ -42,7 +42,7 @@ const (
 
 	MaxTextFieldValueLength = 32 * 1024
 
-	SealOnExitFracSizePercent = 20 // Percent of the max frac size, above which the fraction is sealed on exit
+	MinSealPercent = 20 // Percent of the max frac size, above which the fraction is sealed on exit
 
 	IngestorMaxInflightBulks = 32
 

frac/active.go

@@ -17,6 +17,7 @@ import (
 	"github.com/ozontech/seq-db/config"
 	"github.com/ozontech/seq-db/consts"
 	"github.com/ozontech/seq-db/frac/common"
+	"github.com/ozontech/seq-db/frac/processor"
 	"github.com/ozontech/seq-db/logger"
 	"github.com/ozontech/seq-db/metric"
 	"github.com/ozontech/seq-db/metric/stopwatch"
@@ -34,10 +35,6 @@ type Active struct {
 
 	BaseFileName string
 
-	useMu    sync.RWMutex
-	suicided bool
-	released bool
-
 	infoMu sync.RWMutex
 	info   *common.Info
 
@@ -266,23 +263,19 @@ func (f *Active) String() string {
 	return fracToString(f, "active")
 }
 
-func (f *Active) DataProvider(ctx context.Context) (DataProvider, func()) {
-	f.useMu.RLock()
-
-	if f.suicided || f.released || f.Info().DocsTotal == 0 { // it is empty active fraction state
-		if f.suicided {
-			metric.CountersTotal.WithLabelValues("fraction_suicided").Inc()
-		}
-		f.useMu.RUnlock()
-		return EmptyDataProvider{}, func() {}
+func (f *Active) Fetch(ctx context.Context, ids []seq.ID) ([][]byte, error) {
+	if f.Info().DocsTotal == 0 { // it is empty active fraction state
+		return nil, nil
 	}
+	return f.createDataProvider(ctx).Fetch(ids)
+}
 
-	// it is ordinary active fraction state
-	dp := f.createDataProvider(ctx)
-	return dp, func() {
-		dp.release()
-		f.useMu.RUnlock()
+func (f *Active) Search(ctx context.Context, params processor.SearchParams) (*seq.QPR, error) {
+	if f.Info().DocsTotal == 0 { // it is empty active fraction state
+		metric.CountersTotal.WithLabelValues("empty_data_provider").Inc()
+		return &seq.QPR{Aggs: make([]seq.AggregatableSamples, len(params.AggQ))}, nil
 	}
+	return f.createDataProvider(ctx).Search(params)
 }
 
 func (f *Active) createDataProvider(ctx context.Context) *activeDataProvider {
@@ -318,10 +311,6 @@ func (f *Active) IsIntersecting(from, to seq.MID) bool {
 }
 
 func (f *Active) Release() {
-	f.useMu.Lock()
-	f.released = true
-	f.useMu.Unlock()
-
 	f.releaseMem()
 
 	if !f.Config.KeepMetaFile {
@@ -334,35 +323,6 @@ func (f *Active) Release() {
 	}
 }
 
-// Offload for [Active] fraction is no-op.
-//
-// Since search within [Active] fraction is too costly (we have to replay the whole index in memory),
-// we decided to support offloading only for [Sealed] fractions.
-func (f *Active) Offload(context.Context, storage.Uploader) (bool, error) {
-	return false, nil
-}
-
-func (f *Active) Suicide() {
-	f.useMu.Lock()
-	released := f.released
-	f.suicided = true
-	f.released = true
-	f.useMu.Unlock()
-
-	if released { // fraction can be suicided after release
-		if f.Config.KeepMetaFile {
-			f.removeMetaFile() // meta was not removed while release
-		}
-		if f.Config.SkipSortDocs {
-			f.removeDocsFiles() // docs was not removed while release
-		}
-	} else { // was not release
-		f.releaseMem()
-		f.removeMetaFile()
-		f.removeDocsFiles()
-	}
-}
-
 func (f *Active) releaseMem() {
 	f.writer.Stop()
 	f.TokenList.Stop()

frac/active_indexer.go

@@ -17,8 +17,6 @@ type ActiveIndexer struct {
 	ch          chan *indexTask
 	chMerge     chan *mergeTask
 	workerCount int
-
-	stopFn func()
 }
 
 type indexTask struct {
@@ -33,12 +31,14 @@ type mergeTask struct {
 	tokenLIDs *TokenLIDs
 }
 
-func NewActiveIndexer(workerCount, chLen int) *ActiveIndexer {
-	return &ActiveIndexer{
+func NewActiveIndexer(workerCount, chLen int) (*ActiveIndexer, func()) {
+	idx := ActiveIndexer{
 		ch:          make(chan *indexTask, chLen),
 		chMerge:     make(chan *mergeTask, chLen),
 		workerCount: workerCount,
 	}
+	stopIdx := idx.start()
+	return &idx, stopIdx
 }
 
 func (ai *ActiveIndexer) Index(frac *Active, metas []byte, wg *sync.WaitGroup, sw *stopwatch.Stopwatch) {
@@ -52,7 +52,7 @@ func (ai *ActiveIndexer) Index(frac *Active, metas []byte, wg *sync.WaitGroup, s
 	m.Stop()
 }
 
-func (ai *ActiveIndexer) Start() {
+func (ai *ActiveIndexer) start() func() {
 	wg := sync.WaitGroup{}
 	wg.Add(ai.workerCount)
 
@@ -71,13 +71,10 @@ func (ai *ActiveIndexer) Start() {
 		}()
 	}
 
-	ai.stopFn = func() {
+	return func() {
 		close(ai.ch)
 		close(ai.chMerge)
-
 		wg.Wait()
-
-		ai.stopFn = nil
 	}
 }
 
@@ -87,12 +84,6 @@ func (ai *ActiveIndexer) mergeWorker() {
 	}
 }
 
-func (ai *ActiveIndexer) Stop() {
-	if ai.stopFn != nil {
-		ai.stopFn()
-	}
-}
-
 var metaDataPool = sync.Pool{
 	New: func() any {
 		return new(MetaData)

frac/fraction.go

@@ -9,21 +9,14 @@ import (
 	"github.com/ozontech/seq-db/frac/common"
 	"github.com/ozontech/seq-db/frac/processor"
 	"github.com/ozontech/seq-db/seq"
-	"github.com/ozontech/seq-db/storage"
 )
 
-type DataProvider interface {
-	Fetch([]seq.ID) ([][]byte, error)
-	Search(processor.SearchParams) (*seq.QPR, error)
-}
-
 type Fraction interface {
 	Info() *common.Info
 	IsIntersecting(from seq.MID, to seq.MID) bool
 	Contains(mid seq.MID) bool
-	DataProvider(context.Context) (DataProvider, func())
-	Offload(ctx context.Context, u storage.Uploader) (bool, error)
-	Suicide()
+	Fetch(context.Context, []seq.ID) ([][]byte, error)
+	Search(context.Context, processor.SearchParams) (*seq.QPR, error)
 }
 
 func fracToString(f Fraction, fracType string) string {

frac/remote.go

@@ -11,12 +11,12 @@ import (
 	"github.com/ozontech/seq-db/cache"
 	"github.com/ozontech/seq-db/consts"
 	"github.com/ozontech/seq-db/frac/common"
+	"github.com/ozontech/seq-db/frac/processor"
 	"github.com/ozontech/seq-db/frac/sealed"
 	"github.com/ozontech/seq-db/frac/sealed/lids"
 	"github.com/ozontech/seq-db/frac/sealed/seqids"
 	"github.com/ozontech/seq-db/frac/sealed/token"
 	"github.com/ozontech/seq-db/logger"
-	"github.com/ozontech/seq-db/metric"
 	"github.com/ozontech/seq-db/seq"
 	"github.com/ozontech/seq-db/storage"
 	"github.com/ozontech/seq-db/storage/s3"
@@ -41,9 +41,6 @@ type Remote struct {
 
 	info *common.Info
 
-	useMu    sync.RWMutex
-	suicided bool
-
 	docsFile   storage.ImmutableFile
 	docsCache  *cache.Cache[[]byte]
 	docsReader storage.DocsReader
@@ -114,37 +111,20 @@ func (f *Remote) Contains(mid seq.MID) bool {
 	return f.info.IsIntersecting(mid, mid)
 }
 
-func (f *Remote) DataProvider(ctx context.Context) (DataProvider, func()) {
-	f.useMu.RLock()
-
-	if f.suicided {
-		metric.CountersTotal.WithLabelValues("fraction_suicided").Inc()
-		f.useMu.RUnlock()
-		return EmptyDataProvider{}, func() {}
-	}
-
-	defer func() {
-		if panicData := recover(); panicData != nil {
-			f.useMu.RUnlock()
-			panic(panicData)
-		}
-	}()
-
-	if err := f.load(); err != nil {
-		logger.Error(
-			"will create empty data provider: cannot load remote fraction",
-			zap.String("fraction", f.Info().Name()),
-			zap.Error(err),
-		)
-		f.useMu.RUnlock()
-		return EmptyDataProvider{}, func() {}
+func (f *Remote) Fetch(ctx context.Context, ids []seq.ID) ([][]byte, error) {
+	dp, err := f.createDataProvider(ctx)
+	if err != nil {
+		return nil, err
 	}
+	return dp.Fetch(ids)
+}
 
-	dp := f.createDataProvider(ctx)
-	return dp, func() {
-		dp.release()
-		f.useMu.RUnlock()
+func (f *Remote) Search(ctx context.Context, params processor.SearchParams) (*seq.QPR, error) {
+	dp, err := f.createDataProvider(ctx)
+	if err != nil {
+		return &seq.QPR{Aggs: make([]seq.AggregatableSamples, len(params.AggQ))}, err
 	}
+	return dp.Search(params)
 }
 
 func (f *Remote) Info() *common.Info {
@@ -155,15 +135,7 @@ func (f *Remote) IsIntersecting(from, to seq.MID) bool {
 	return f.info.IsIntersecting(from, to)
 }
 
-func (f *Remote) Offload(context.Context, storage.Uploader) (bool, error) {
-	panic("BUG: remote fraction cannot be offloaded")
-}
-
 func (f *Remote) Suicide() {
-	f.useMu.Lock()
-	f.suicided = true
-	f.useMu.Unlock()
-
 	util.MustRemoveFileByPath(f.BaseFileName + consts.RemoteFractionSuffix)
 
 	f.docsCache.Release()
@@ -189,7 +161,15 @@ func (f *Remote) String() string {
 	return fracToString(f, "remote")
 }
 
-func (f *Remote) createDataProvider(ctx context.Context) *sealedDataProvider {
+func (f *Remote) createDataProvider(ctx context.Context) (*sealedDataProvider, error) {
+	if err := f.load(); err != nil {
+		logger.Error(
+			"will create empty data provider: cannot load remote fraction",
+			zap.String("fraction", f.Info().Name()),
+			zap.Error(err),
+		)
+		return nil, err
+	}
 	return &sealedDataProvider{
 		ctx:              ctx,
 		info:             f.info,
@@ -210,7 +190,7 @@ func (f *Remote) createDataProvider(ctx context.Context) *sealedDataProvider {
 			&f.blocksData.IDsTable,
 			f.info.BinaryDataVer,
 		),
-	}
+	}, nil
 }
 
 func (f *Remote) load() error {