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

No handover (#378)

Browse Source 
Co-authored-by: Alexey Sharp <alexeysharp@Alexeys-iMac.local>
This commit is contained in:
ledgerwatch 2022-03-18 16:09:57 +00:00 committed by GitHub
parent f93ea948d0
commit ff3e074549
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 10 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
aggregator/aggregator.go
View File

@ -164,7 +164,6 @@ type Aggregator struct {
changesets bool // Whether to generate changesets (off by default) changesets bool // Whether to generate changesets (off by default)
commitments bool // Whether to calculate commitments commitments bool // Whether to calculate commitments
aggChannel chan *AggregationTask aggChannel chan *AggregationTask
aggBackCh chan struct{} // Channel for acknoledgement of AggregationTask
aggError chan error aggError chan error
aggWg sync.WaitGroup aggWg sync.WaitGroup
mergeChannel chan struct{} mergeChannel chan struct{}
@ -948,7 +947,6 @@ func NewAggregator(diffDir string, unwindLimit uint64, aggregationStep uint64, c
keccak: sha3.NewLegacyKeccak256(), keccak: sha3.NewLegacyKeccak256(),
hph: commitment.NewHexPatriciaHashed(length.Addr, nil, nil, nil, nil, nil), hph: commitment.NewHexPatriciaHashed(length.Addr, nil, nil, nil, nil, nil),
aggChannel: make(chan *AggregationTask), aggChannel: make(chan *AggregationTask),
aggBackCh: make(chan struct{}),
aggError: make(chan error, 1), aggError: make(chan error, 1),
mergeChannel: make(chan struct{}, 1), mergeChannel: make(chan struct{}, 1),
mergeError: make(chan error, 1), mergeError: make(chan error, 1),
@ -1183,17 +1181,6 @@ func removeFiles(fType FileType, diffDir string, toRemove []*byEndBlockItem) err
func (a *Aggregator) backgroundAggregation() { func (a *Aggregator) backgroundAggregation() {
defer a.aggWg.Done() defer a.aggWg.Done()
for aggTask := range a.aggChannel { for aggTask := range a.aggChannel {
typesLimit := Commitment
if a.commitments {
typesLimit = AccountHistory
}
for fType := FirstType; fType < typesLimit; fType++ {
if fType < NumberOfStateTypes {
a.updateArch(aggTask.bt[fType], fType, uint32(aggTask.blockTo))
}
a.addLocked(fType, &byEndBlockItem{startBlock: aggTask.blockFrom, endBlock: aggTask.blockTo, tree: aggTask.bt[fType]})
}
a.aggBackCh <- struct{}{}
if a.changesets { if a.changesets {
if historyD, historyI, bitmapD, bitmapI, err := aggTask.changes[Account].produceChangeSets(aggTask.blockFrom, aggTask.blockTo, AccountHistory, AccountBitmap); err == nil { if historyD, historyI, bitmapD, bitmapI, err := aggTask.changes[Account].produceChangeSets(aggTask.blockFrom, aggTask.blockTo, AccountHistory, AccountBitmap); err == nil {
var historyItem = &byEndBlockItem{startBlock: aggTask.blockFrom, endBlock: aggTask.blockTo} var historyItem = &byEndBlockItem{startBlock: aggTask.blockFrom, endBlock: aggTask.blockTo}
@ -1259,6 +1246,10 @@ func (a *Aggregator) backgroundAggregation() {
return return
} }
} }
typesLimit := Commitment
if a.commitments {
typesLimit = AccountHistory
}
for fType := FirstType; fType < typesLimit; fType++ { for fType := FirstType; fType < typesLimit; fType++ {
var err error var err error
if err = aggTask.changes[fType].closeFiles(); err != nil { if err = aggTask.changes[fType].closeFiles(); err != nil {
@ -2902,8 +2893,13 @@ func (w *Writer) aggregateUpto(blockFrom, blockTo uint64) error {
t = time.Now() t = time.Now()
// At this point, all the changes are gathered in 4 B-trees (accounts, code, storage and commitment) and removed from the database // At this point, all the changes are gathered in 4 B-trees (accounts, code, storage and commitment) and removed from the database
// What follows can be done in the 1st background goroutine // What follows can be done in the 1st background goroutine
for fType := FirstType; fType < typesLimit; fType++ {
if fType < NumberOfStateTypes {
w.a.updateArch(aggTask.bt[fType], fType, uint32(aggTask.blockTo))
}
w.a.addLocked(fType, &byEndBlockItem{startBlock: aggTask.blockFrom, endBlock: aggTask.blockTo, tree: aggTask.bt[fType]})
}
w.a.aggChannel <- &aggTask w.a.aggChannel <- &aggTask
<-w.a.aggBackCh // Waiting for the B-tree based items have been added
handoverTime := time.Since(t) handoverTime := time.Since(t)
if handoverTime > time.Second { if handoverTime > time.Second {
log.Info("Long handover to background aggregation", "from", blockFrom, "to", blockTo, "composition", aggTime, "handover", time.Since(t)) log.Info("Long handover to background aggregation", "from", blockFrom, "to", blockTo, "composition", aggTime, "handover", time.Since(t))