mirror/erigon-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/erigon-pulse.git synced 2025-01-07 11:32:20 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

create .seg in tmpdir

Browse Source
This commit is contained in:
alex.sharov 2022-01-09 14:49:56 +07:00
parent 0d5d8975d9
commit 8bc0f26a49
1 changed files with 11 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
compress/parallel_compress.go
View File

@ -11,6 +11,7 @@ import (
"fmt" "fmt"
"io" "io"
"os" "os"
"path/filepath"
"runtime" "runtime"
"sort" "sort"
"strconv" "strconv"
@ -31,6 +32,10 @@ import (
const minPatternScore = 1024 const minPatternScore = 1024
func Compress(ctx context.Context, logPrefix, tmpFilePath, segmentFilePath string, workers int) error { func Compress(ctx context.Context, logPrefix, tmpFilePath, segmentFilePath string, workers int) error {
tmpDir, _ := filepath.Split(tmpFilePath)
_, fileName := filepath.Split(segmentFilePath)
tmpSegmentFilePath := filepath.Join(tmpDir, fileName)
logEvery := time.NewTicker(20 * time.Second) logEvery := time.NewTicker(20 * time.Second)
defer logEvery.Stop() defer logEvery.Stop()
@ -39,7 +44,6 @@ func Compress(ctx context.Context, logPrefix, tmpFilePath, segmentFilePath strin
var superstring []byte var superstring []byte
// Collector for dictionary words (sorted by their score) // Collector for dictionary words (sorted by their score)
tmpDir := ""
ch := make(chan []byte, workers) ch := make(chan []byte, workers)
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Add(workers) wg.Add(workers)
@ -92,7 +96,11 @@ func Compress(ctx context.Context, logPrefix, tmpFilePath, segmentFilePath strin
return err return err
} }
if err := reducedict(logPrefix, tmpFilePath, dictPath, segmentFilePath); err != nil { if err := reducedict(logPrefix, tmpFilePath, dictPath, tmpSegmentFilePath, tmpDir); err != nil {
return err
}
if err := os.Rename(tmpSegmentFilePath, segmentFilePath); err != nil {
return err return err
} }
return nil return nil
@ -267,7 +275,7 @@ func reduceDictWorker(inputCh chan []byte, completion *sync.WaitGroup, trie *pat
} }
// reduceDict reduces the dictionary by trying the substitutions and counting frequency for each word // reduceDict reduces the dictionary by trying the substitutions and counting frequency for each word
func reducedict(logPrefix, tmpFilePath, dictPath, segmentFilePath string) error { func reducedict(logPrefix, tmpFilePath, dictPath, segmentFilePath, tmpDir string) error {
logEvery := time.NewTicker(20 * time.Second) logEvery := time.NewTicker(20 * time.Second)
defer logEvery.Stop() defer logEvery.Stop()
@ -289,7 +297,6 @@ func reducedict(logPrefix, tmpFilePath, dictPath, segmentFilePath string) error
return err return err
} }
log.Info(fmt.Sprintf("[%s] dictionary file parsed", logPrefix), "entries", len(code2pattern)) log.Info(fmt.Sprintf("[%s] dictionary file parsed", logPrefix), "entries", len(code2pattern))
tmpDir := ""
ch := make(chan []byte, 10000) ch := make(chan []byte, 10000)
inputSize, outputSize := atomic2.NewUint64(0), atomic2.NewUint64(0) inputSize, outputSize := atomic2.NewUint64(0), atomic2.NewUint64(0)
var wg sync.WaitGroup var wg sync.WaitGroup