Improve staged stream sync, fix the devnet node stuck issue (#4467)
* add error desc in validateNewBlock to help to identify validation issues * let stream sync downloader continues loop even if error occured * pass consensus to stream sync through downloader * add last mile functions to stream sync * add if to check invalid block revert * add last mile stage to stream sync * goimports * improve stream sync downloader loop to block redundant calls * move startSyncing out of the shortrange loop * goimports * fix sync loop go routine * remove extra log * add debug mode to stream sync help debugging syncing issues * fix stream sync loop channels * add streamFailed function to short range helper to avoid removing of the healthy streams * remove execution of stage bodies, stage lastmile, stage short range and stage state for epoch chain * refactor stage epoch * add debug logs * goimports * add a few debug log to stage short range * doSync returns estimated height as well * only switch to short range if the current block number is very close to the chain current height * stream sync gets UseMemDb from config file * goimports * only flag failed streams rather than removing them in stream sync * if stage blocks progress behind current head, remove block cache * add rollback to short range * refactor stage last mile blocks, add roll back to this stage * improve addConsensusLastMile * goimports * fix log spell error * improve revert function, no need to revert if hashes are empty * fix switch to short range by removing extra condition * add donC chan size * refactor downloader loop mechanism * use atomic flag rather than done channel in downloader loop * no need for fail stream in epoch sync * ignore context timeout * add mux lock to get access to last mile blocks * remove atomic flag for downloader loop * a few improvements on staged sync, check addedBn before switch to short range * goimports * fix consensus catchup issue * fix panic issue from runnig sync loop while stream sync is runing * goimports * add two more logs to staged sync * remove extra debug logs, add more file logs for stream sync * add comment for DebugMode * improve the byte comparison for getBlockFromLastMileBlocksByParentHash functionpull/4487/head
Gheis Mohammadi
1 year ago
committed by
GitHub
parent
764474acae
commit
b093fea169
@ -0,0 +1,109 @@ |
||||
package stagedstreamsync |
||||
|
||||
import ( |
||||
"context" |
||||
|
||||
"github.com/ethereum/go-ethereum/common" |
||||
"github.com/harmony-one/harmony/core" |
||||
"github.com/harmony-one/harmony/shard" |
||||
"github.com/ledgerwatch/erigon-lib/kv" |
||||
) |
||||
|
||||
type StageLastMile struct { |
||||
configs StageLastMileCfg |
||||
} |
||||
|
||||
type StageLastMileCfg struct { |
||||
ctx context.Context |
||||
bc core.BlockChain |
||||
db kv.RwDB |
||||
} |
||||
|
||||
func NewStageLastMile(cfg StageLastMileCfg) *StageLastMile { |
||||
return &StageLastMile{ |
||||
configs: cfg, |
||||
} |
||||
} |
||||
|
||||
func NewStageLastMileCfg(ctx context.Context, bc core.BlockChain, db kv.RwDB) StageLastMileCfg { |
||||
return StageLastMileCfg{ |
||||
ctx: ctx, |
||||
bc: bc, |
||||
db: db, |
||||
} |
||||
} |
||||
|
||||
func (lm *StageLastMile) Exec(ctx context.Context, firstCycle bool, invalidBlockRevert bool, s *StageState, reverter Reverter, tx kv.RwTx) (err error) { |
||||
|
||||
// no need to download the last mile blocks if we are redoing the stages because of bad block
|
||||
if invalidBlockRevert { |
||||
return nil |
||||
} |
||||
|
||||
// shouldn't execute for epoch chain
|
||||
if lm.configs.bc.ShardID() == shard.BeaconChainShardID && !s.state.isBeaconNode { |
||||
return nil |
||||
} |
||||
|
||||
bc := lm.configs.bc |
||||
|
||||
// update last mile blocks if any
|
||||
parentHash := bc.CurrentBlock().Hash() |
||||
var hashes []common.Hash |
||||
for { |
||||
block := s.state.getBlockFromLastMileBlocksByParentHash(parentHash) |
||||
if block == nil { |
||||
break |
||||
} |
||||
err = s.state.UpdateBlockAndStatus(block, bc, false) |
||||
if err != nil { |
||||
s.state.RollbackLastMileBlocks(ctx, hashes) |
||||
return err |
||||
} |
||||
hashes = append(hashes, block.Hash()) |
||||
parentHash = block.Hash() |
||||
} |
||||
s.state.purgeLastMileBlocksFromCache() |
||||
|
||||
return nil |
||||
} |
||||
|
||||
func (lm *StageLastMile) Revert(ctx context.Context, firstCycle bool, u *RevertState, s *StageState, tx kv.RwTx) (err error) { |
||||
useInternalTx := tx == nil |
||||
if useInternalTx { |
||||
tx, err = lm.configs.db.BeginRw(lm.configs.ctx) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
defer tx.Rollback() |
||||
} |
||||
|
||||
if err = u.Done(tx); err != nil { |
||||
return err |
||||
} |
||||
|
||||
if useInternalTx { |
||||
if err = tx.Commit(); err != nil { |
||||
return err |
||||
} |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
func (lm *StageLastMile) CleanUp(ctx context.Context, firstCycle bool, p *CleanUpState, tx kv.RwTx) (err error) { |
||||
useInternalTx := tx == nil |
||||
if useInternalTx { |
||||
tx, err = lm.configs.db.BeginRw(lm.configs.ctx) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
defer tx.Rollback() |
||||
} |
||||
|
||||
if useInternalTx { |
||||
if err = tx.Commit(); err != nil { |
||||
return err |
||||
} |
||||
} |
||||
return nil |
||||
} |
||||
Loading…
Reference in new issue