e3: parallel miss idx (#728)

erigontech · Nov 7, 2022 · bfd3996 · bfd3996
1 parent 4b0ff0a
commit bfd3996
Show file tree

Hide file tree

Showing 3 changed files with 68 additions and 74 deletions.
diff --git a/state/domain.go b/state/domain.go
@@ -796,8 +796,6 @@ func (d *Domain) BuildMissedIndices(ctx context.Context, sem *semaphore.Weighted
 }
 
 func buildIndex(ctx context.Context, d *compress.Decompressor, idxPath, tmpdir string, count int, values bool) (*recsplit.Index, error) {
-	_, fName := filepath.Split(idxPath)
-	log.Debug("[snapshots] build idx", "file", fName)
 	var rs *recsplit.RecSplit
 	var err error
 	if rs, err = recsplit.NewRecSplit(recsplit.RecSplitArgs{

diff --git a/state/history.go b/state/history.go
@@ -271,52 +271,49 @@ func (h *History) BuildMissedIndices(ctx context.Context, sem *semaphore.Weighte
 	}
 
 	missedFiles := h.missedIdxFiles()
-	//errs := make(chan error, len(missedFiles))
-	//wg := sync.WaitGroup{}
+	errs := make(chan error, len(missedFiles))
+	wg := sync.WaitGroup{}
 
 	for _, item := range missedFiles {
-		//if err := sem.Acquire(ctx, 1); err != nil {
-		//	errs <- err
-		//	break
-		//}
-		//wg.Add(1)
-		//go func(item *filesItem) {
-		//	defer sem.Release(1)
-		//	defer wg.Done()
-
-		search := &filesItem{startTxNum: item.startTxNum, endTxNum: item.endTxNum}
-		iiItem, ok := h.InvertedIndex.files.Get(search)
-		if !ok {
-			return nil
+		if err := sem.Acquire(ctx, 1); err != nil {
+			errs <- err
+			break
 		}
+		wg.Add(1)
+		go func(item *filesItem) {
+			defer sem.Release(1)
+			defer wg.Done()
 
-		fromStep, toStep := item.startTxNum/h.aggregationStep, item.endTxNum/h.aggregationStep
-		fName := fmt.Sprintf("%s.%d-%d.vi", h.filenameBase, fromStep, toStep)
-		idxPath := filepath.Join(h.dir, fName)
-		log.Info("[snapshots] build idx", "file", fName)
-		count, err := iterateForVi(item, iiItem, h.compressVals, func(v []byte) error { return nil })
+			search := &filesItem{startTxNum: item.startTxNum, endTxNum: item.endTxNum}
+			iiItem, ok := h.InvertedIndex.files.Get(search)
+			if !ok {
+				return
+			}
+
+			fromStep, toStep := item.startTxNum/h.aggregationStep, item.endTxNum/h.aggregationStep
+			fName := fmt.Sprintf("%s.%d-%d.vi", h.filenameBase, fromStep, toStep)
+			idxPath := filepath.Join(h.dir, fName)
+			log.Info("[snapshots] build idx", "file", fName)
+			count, err := iterateForVi(item, iiItem, h.compressVals, func(v []byte) error { return nil })
+			if err != nil {
+				errs <- err
+			}
+			errs <- buildVi(item, iiItem, idxPath, h.tmpdir, count, false /* values */, h.compressVals)
+		}(item)
+	}
+	go func() {
+		wg.Wait()
+		close(errs)
+	}()
+	var lastError error
+	for err := range errs {
 		if err != nil {
-			return err
+			lastError = err
 		}
-		if err := buildVi(item, iiItem, idxPath, h.tmpdir, count, false /* values */, h.compressVals); err != nil {
-			return err
-		}
-		//errs <- buildVi(item, iiItem, idxPath, h.tmpdir, count, false /* values */, h.compressVals)
-		//}(item)
-	}
-	//go func() {
-	//	wg.Wait()
-	//	close(errs)
-	//}()
-	//var lastError error
-	//for err := range errs {
-	//	if err != nil {
-	//		lastError = err
-	//	}
-	//}
-	//if lastError != nil {
-	//	return lastError
-	//}
+	}
+	if lastError != nil {
+		return lastError
+	}
 
 	return h.openFiles()
 }

diff --git a/state/inverted_index.go b/state/inverted_index.go
@@ -196,41 +196,40 @@ func (ii *InvertedIndex) missedIdxFiles() (l []*filesItem) {
 // BuildMissedIndices - produce .efi/.vi/.kvi from .ef/.v/.kv
 func (ii *InvertedIndex) BuildMissedIndices(ctx context.Context, sem *semaphore.Weighted) (err error) {
 	missedFiles := ii.missedIdxFiles()
-	//errs := make(chan error, len(missedFiles))
-	//wg := sync.WaitGroup{}
+	errs := make(chan error, len(missedFiles))
+	wg := sync.WaitGroup{}
 	for _, item := range missedFiles {
-		//if err := sem.Acquire(ctx, 1); err != nil {
-		//	errs <- err
-		//	break
-		//}
-		//wg.Add(1)
-		//go func(item *filesItem) {
-		//	defer sem.Release(1)
-		//	defer wg.Done()
-		fromStep, toStep := item.startTxNum/ii.aggregationStep, item.endTxNum/ii.aggregationStep
-		fName := fmt.Sprintf("%s.%d-%d.efi", ii.filenameBase, fromStep, toStep)
-		idxPath := filepath.Join(ii.dir, fName)
-		log.Info("[snapshots] build idx", "file", fName)
-		_, err := buildIndex(ctx, item.decompressor, idxPath, ii.tmpdir, item.decompressor.Count()/2, false)
+		if err := sem.Acquire(ctx, 1); err != nil {
+			errs <- err
+			break
+		}
+		wg.Add(1)
+		go func(item *filesItem) {
+			defer sem.Release(1)
+			defer wg.Done()
+			fromStep, toStep := item.startTxNum/ii.aggregationStep, item.endTxNum/ii.aggregationStep
+			fName := fmt.Sprintf("%s.%d-%d.efi", ii.filenameBase, fromStep, toStep)
+			idxPath := filepath.Join(ii.dir, fName)
+			log.Info("[snapshots] build idx", "file", fName)
+			_, err := buildIndex(ctx, item.decompressor, idxPath, ii.tmpdir, item.decompressor.Count()/2, false)
+			if err != nil {
+				errs <- err
+			}
+		}(item)
+	}
+	go func() {
+		wg.Wait()
+		close(errs)
+	}()
+	var lastError error
+	for err := range errs {
 		if err != nil {
-			return err
+			lastError = err
 		}
-		//errs <- err
-		//}(item)
-	}
-	//go func() {
-	//	wg.Wait()
-	//	close(errs)
-	//}()
-	//var lastError error
-	//for err := range errs {
-	//	if err != nil {
-	//		lastError = err
-	//	}
-	//}
-	//if lastError != nil {
-	//	return lastError
-	//}
+	}
+	if lastError != nil {
+		return lastError
+	}
 	return ii.openFiles()
 }