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

Fix last chunk order (#683)

Browse Source 
* fix commit order

* fix lint
This commit is contained in:
b00ris 2020-06-21 17:13:17 +03:00 committed by GitHub
parent 69a20de5cc
commit 7ac009a7da
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 19 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
common/etl/collector.go
View File

@ -132,16 +132,10 @@ func loadFilesIntoBucket(db ethdb.Database, bucket []byte, providers []dataProvi
return err
}
}
var currentKeyStr string
if len(k) < 4 {
currentKeyStr = fmt.Sprintf("%x", k)
} else {
currentKeyStr = fmt.Sprintf("%x...", k[:4])
}
batchCh <- struct {
db ethdb.DbWithPendingMutations
currentKey string
}{db: batch, currentKey: currentKeyStr}
}{db: batch, currentKey: makeCurrentKeyStr(k)}
batch = getBatch()
}
return nil
@ -173,9 +167,12 @@ func loadFilesIntoBucket(db ethdb.Database, bucket []byte, providers []dataProvi
return err
}
}
_, err := batch.Commit()
batchCh <- struct {
db ethdb.DbWithPendingMutations
currentKey string
}{db: batch, currentKey: makeCurrentKeyStr(nil)}
return err
return nil
})
wg.Go(func() error {
commit := func(batchToCommit struct {
@ -218,3 +215,15 @@ func loadFilesIntoBucket(db ethdb.Database, bucket []byte, providers []dataProvi
return wg.Wait()
}
func makeCurrentKeyStr(k []byte) string {
var currentKeyStr string
if k == nil {
currentKeyStr = "final"
} else if len(k) < 4 {
currentKeyStr = fmt.Sprintf("%x", k)
} else {
currentKeyStr = fmt.Sprintf("%x...", k[:4])
}
return currentKeyStr
}

10
common/etl/dataprovider.go
View File

@ -50,17 +50,9 @@ func FlushToDisk(encoder Encoder, currentKey []byte, b Buffer, datadir string) (
}
var m runtime.MemStats
runtime.ReadMemStats(&m)
var currentKeyStr string
if currentKey == nil {
currentKeyStr = "final"
} else if len(currentKey) < 4 {
currentKeyStr = fmt.Sprintf("%x", currentKey)
} else {
currentKeyStr = fmt.Sprintf("%x...", currentKey[:4])
}
log.Info(
"Flushed buffer file",
"current key", currentKeyStr,
"current key", makeCurrentKeyStr(currentKey),
"name", bufferFile.Name(),
"alloc", common.StorageSize(m.Alloc), "sys", common.StorageSize(m.Sys), "numGC", int(m.NumGC))
b.Reset()

1
common/etl/etl_test.go
View File

@ -176,7 +176,6 @@ func TestTransformOnLoadCommitCustomBatchSize(t *testing.T) {
},
)
assert.Nil(t, err)
fmt.Println(numberOfCalls)
compareBuckets(t, db, sourceBucket, destBucket, nil)
assert.Equal(t, 21, numberOfCalls)