refactor(fracmanager): using fifo queues of fractions (#268)

Author: eguguchkin

cmd/seq-db/seq-db.go

@@ -259,6 +259,7 @@ func startStore(
 			MaintenanceDelay:  0,
 			CacheGCDelay:      0,
 			CacheCleanupDelay: 0,
+			MinSealFracSize:   uint64(cfg.Storage.FracSize) * consts.DefaultMinSealPercent / 100,
 			SealParams: common.SealParams{
 				IDsZstdLevel:           cfg.Compression.SealedZstdCompressionLevel,
 				LIDsZstdLevel:          cfg.Compression.SealedZstdCompressionLevel,

frac/active.go

@@ -35,10 +35,6 @@ type Active struct {
 
 	BaseFileName string
 
-	useMu    sync.RWMutex
-	suicided bool
-	released bool
-
 	infoMu sync.RWMutex
 	info   *common.Info
 
@@ -270,40 +266,26 @@ func (f *Active) String() string {
 }
 
 func (f *Active) Fetch(ctx context.Context, ids []seq.ID) ([][]byte, error) {
-	dp, release := f.DataProvider(ctx)
-	defer release()
-	if dp == nil {
-		return EmptyFraction.Fetch(ctx, ids)
+	if f.Info().DocsTotal == 0 { // it is empty active fraction state
+		return nil, nil
 	}
+
+	dp := f.createDataProvider(ctx)
+	defer dp.release()
+
 	return dp.Fetch(ids)
 }
 
 func (f *Active) Search(ctx context.Context, params processor.SearchParams) (*seq.QPR, error) {
-	dp, release := f.DataProvider(ctx)
-	defer release()
-	if dp == nil {
-		return EmptyFraction.Search(ctx, params)
+	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 dp.Search(params)
-}
-
-func (f *Active) DataProvider(ctx context.Context) (*activeDataProvider, 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 nil, func() {}
-	}
-
-	// it is ordinary active fraction state
 	dp := f.createDataProvider(ctx)
-	return dp, func() {
-		dp.release()
-		f.useMu.RUnlock()
-	}
+	defer dp.release()
+
+	return dp.Search(params)
 }
 
 func (f *Active) createDataProvider(ctx context.Context) *activeDataProvider {
@@ -339,49 +321,24 @@ 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 {
-		f.removeMetaFile()
+		util.RemoveFile(f.metaFile.Name())
 	}
 
 	if !f.Config.SkipSortDocs {
 		// we use sorted docs in sealed fraction so we can remove original docs of active fraction
-		f.removeDocsFiles()
+		util.RemoveFile(f.docsFile.Name())
 	}
 }
 
-// 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()
-	}
+	f.releaseMem()
+
+	util.RemoveFile(f.metaFile.Name())
+	util.RemoveFile(f.docsFile.Name())
+	util.RemoveFile(f.BaseFileName + consts.SdocsFileSuffix)
 }
 
 func (f *Active) releaseMem() {
@@ -394,24 +351,12 @@ func (f *Active) releaseMem() {
 	if err := f.metaFile.Close(); err != nil {
 		logger.Error("can't close meta file", zap.String("frac", f.BaseFileName), zap.Error(err))
 	}
+	if err := f.docsFile.Close(); err != nil {
+		logger.Error("can't close docs file", zap.String("frac", f.BaseFileName), zap.Error(err))
+	}
 
 	f.RIDs = nil
 	f.MIDs = nil
 	f.TokenList = nil
 	f.DocsPositions = nil
 }
-
-func (f *Active) removeDocsFiles() {
-	if err := f.docsFile.Close(); err != nil {
-		logger.Error("can't close docs file", zap.String("frac", f.BaseFileName), zap.Error(err))
-	}
-	if err := os.Remove(f.docsFile.Name()); err != nil {
-		logger.Error("can't delete docs file", zap.String("frac", f.BaseFileName), zap.Error(err))
-	}
-}
-
-func (f *Active) removeMetaFile() {
-	if err := os.Remove(f.metaFile.Name()); err != nil {
-		logger.Error("can't delete metas file", zap.String("frac", f.BaseFileName), zap.Error(err))
-	}
-}

frac/active_indexer.go

@@ -20,8 +20,6 @@ type ActiveIndexer struct {
 	chMerge     chan *mergeTask
 	bulkStats   *BulkStatsCollector
 	workerCount int
-
-	stopFn func()
 }
 
 type indexTask struct {
@@ -36,13 +34,15 @@ 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,
 		bulkStats:   NewBulkStatsCollector(5*time.Second, chLen),
 	}
+	stopIdx := idx.start()
+	return &idx, stopIdx
 }
 
 func (ai *ActiveIndexer) Index(frac *Active, metas []byte, wg *sync.WaitGroup, sw *stopwatch.Stopwatch) {
@@ -56,7 +56,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)
 
@@ -75,15 +75,11 @@ func (ai *ActiveIndexer) Start() {
 		}()
 	}
 
-	ai.stopFn = func() {
+	return func() {
 		close(ai.ch)
 		close(ai.chMerge)
-
 		wg.Wait()
-
 		ai.bulkStats.Stop()
-
-		ai.stopFn = nil
 	}
 }
 
@@ -93,12 +89,6 @@ func (ai *ActiveIndexer) mergeWorker() {
 	}
 }
 
-func (ai *ActiveIndexer) Stop() {
-	if ai.stopFn != nil {
-		ai.stopFn()
-	}
-}
-
 var metaDataPool = sync.Pool{
 	New: func() any {
 		return new(indexer.MetaData)

frac/active_indexer_test.go

@@ -76,9 +76,8 @@ func getTestProcessor() *indexer.Processor {
 
 func BenchmarkIndexer(b *testing.B) {
 	logger.SetLevel(zapcore.FatalLevel)
-	idx := NewActiveIndexer(8, 8)
-	idx.Start()
-	defer idx.Stop()
+	idx, stop := NewActiveIndexer(8, 8)
+	defer stop()
 
 	allLogs, err := readFileAllAtOnce(filepath.Join(common.TestDataDir, "k8s.logs"))
 	readers := splitLogsToBulks(allLogs, 1000)

frac/empty.go

@@ -13,7 +13,6 @@ import (
 	"github.com/ozontech/seq-db/frac/processor"
 	"github.com/ozontech/seq-db/metric"
 	"github.com/ozontech/seq-db/seq"
-	"github.com/ozontech/seq-db/storage"
 )
 
 type Fraction interface {
@@ -22,8 +21,6 @@ type Fraction interface {
 	Contains(mid seq.MID) bool
 	Fetch(context.Context, []seq.ID) ([][]byte, error)
 	Search(context.Context, processor.SearchParams) (*seq.QPR, error)
-	Offload(ctx context.Context, u storage.Uploader) (bool, error)
-	Suicide()
 }
 
 var (

frac/fraction.go

@@ -42,9 +42,8 @@ func TestConcurrentAppendAndQuery(t *testing.T) {
 	fracPath := filepath.Join(tmpDir, "test_fraction")
 	defer test_common.RemoveDir(fracPath)
 
-	activeIndexer := NewActiveIndexer(numIndexWorkers, 1000)
-	activeIndexer.Start()
-	defer activeIndexer.Stop()
+	activeIndexer, stop := NewActiveIndexer(numIndexWorkers, 1000)
+	defer stop()
 
 	active := NewActive(
 		fracPath,