mirror/erigon-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/erigon-pulse.git synced 2024-12-28 14:47:16 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

cancel compress (#362)

Browse Source
This commit is contained in:
Alex Sharov 2022-03-12 16:34:58 +07:00 committed by GitHub
parent c0fcdabf91
commit c1f1365f92
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 7 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
compress/compress.go
View File

@ -168,7 +168,7 @@ func (c *Compressor) Compress() error {
}
defer os.Remove(c.tmpOutFilePath)
if err := reducedict(c.trace, c.logPrefix, c.tmpOutFilePath, c.tmpDir, c.uncompressedFile, c.workers, db); err != nil {
if err := reducedict(c.ctx, c.trace, c.logPrefix, c.tmpOutFilePath, c.uncompressedFile, c.workers, db); err != nil {
return err
}

7
compress/parallel_compress.go
View File

@ -238,7 +238,7 @@ func (cq *CompressionQueue) Pop() interface{} {
}
// reduceDict reduces the dictionary by trying the substitutions and counting frequency for each word
func reducedict(trace bool, logPrefix, segmentFilePath, tmpDir string, datFile *DecompressedFile, workers int, dictBuilder *DictionaryBuilder) error {
func reducedict(ctx context.Context, trace bool, logPrefix, segmentFilePath string, datFile *DecompressedFile, workers int, dictBuilder *DictionaryBuilder) error {
logEvery := time.NewTicker(20 * time.Second)
defer logEvery.Stop()
@ -306,6 +306,11 @@ func reducedict(trace bool, logPrefix, segmentFilePath, tmpDir string, datFile *
var inCount, outCount, emptyWordsCount uint64 // Counters words sent to compression and returned for compression
var numBuf [binary.MaxVarintLen64]byte
if err = datFile.ForEach(func(v []byte, compression bool) error {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
if workers > 1 {
// take processed words in non-blocking way and push them to the queue
outer: