prysm-pulse/beacon-chain/sync/initial-sync/blocks_queue_utils.go
Nishant Das caf9bdbc6f
Use Block Interface Across Prysm (#8918)
* commit initial work

* checkpoint current work

* gaz

* checkpoint

* req/resp changes

* initial-sync

* finally works

* fix error

* fix bugs

* fix issue

* fix issues

* fix refs

* tests

* more text fixes

* more text fixes

* more text fixes

* fix tests

* fix tests

* tests

* finally fix builds

* finally

* comments

* fix fuzz

* share common library

* fix

* fix

* add in more defensive nil checks

* add in more defensive nil checks

* imports

* Apply suggestions from code review

Co-authored-by: terence tsao <terence@prysmaticlabs.com>

* Apply suggestions from code review

Co-authored-by: terence tsao <terence@prysmaticlabs.com>

* Update shared/interfaces/block_interface.go

Co-authored-by: terence tsao <terence@prysmaticlabs.com>

* Update shared/interfaces/block_wrapper.go

Co-authored-by: terence tsao <terence@prysmaticlabs.com>

* Update shared/interfaces/block_interface.go

Co-authored-by: terence tsao <terence@prysmaticlabs.com>

* imports

* fix bad changes

* fix

* terence's review

* terence's review

* fmt

* Update beacon-chain/rpc/beacon/blocks.go

Co-authored-by: Radosław Kapka <rkapka@wp.pl>

* fix tests

* fix

* fix all tests

Co-authored-by: terence tsao <terence@prysmaticlabs.com>
Co-authored-by: Radosław Kapka <rkapka@wp.pl>
2021-05-26 16:19:54 +00:00

76 lines
2.6 KiB
Go

package initialsync
import (
"context"
"errors"
types "github.com/prysmaticlabs/eth2-types"
)
// resetWithBlocks removes all state machines, then re-adds enough machines to contain all provided
// blocks (machines are set into stateDataParsed state, so that their content is immediately
// consumable). It is assumed that blocks come in an ascending order.
func (q *blocksQueue) resetFromFork(ctx context.Context, fork *forkData) error {
if fork == nil {
return errors.New("nil fork data")
}
if len(fork.blocks) == 0 {
return errors.New("no blocks to reset from")
}
firstBlock := fork.blocks[0].Block()
if firstBlock == nil || firstBlock.IsNil() {
return errors.New("invalid first block in fork data")
}
blocksPerRequest := q.blocksFetcher.blocksPerSecond
if err := q.smm.removeAllStateMachines(); err != nil {
return err
}
fsm := q.smm.addStateMachine(firstBlock.Slot())
fsm.pid = fork.peer
fsm.blocks = fork.blocks
fsm.state = stateDataParsed
// The rest of machines are in skipped state.
startSlot := firstBlock.Slot().Add(uint64(len(fork.blocks)))
for i := startSlot; i < startSlot.Add(blocksPerRequest*(lookaheadSteps-1)); i += types.Slot(blocksPerRequest) {
fsm := q.smm.addStateMachine(i)
fsm.state = stateSkipped
}
return nil
}
// resetFromSlot removes all state machines, and re-adds them starting with a given slot.
// The last machine added relies on calculated non-skipped slot (to allow FSMs to jump over
// long periods with skipped slots).
func (q *blocksQueue) resetFromSlot(ctx context.Context, startSlot types.Slot) error {
// Shift start position of all the machines except for the last one.
blocksPerRequest := q.blocksFetcher.blocksPerSecond
if err := q.smm.removeAllStateMachines(); err != nil {
return err
}
for i := startSlot; i < startSlot.Add(blocksPerRequest*(lookaheadSteps-1)); i += types.Slot(blocksPerRequest) {
q.smm.addStateMachine(i)
}
// Replace the last (currently activated) state machine to start with best known non-skipped slot.
nonSkippedSlot, err := q.blocksFetcher.nonSkippedSlotAfter(ctx, startSlot.Add(blocksPerRequest*(lookaheadSteps-1)-1))
if err != nil {
return err
}
if q.mode == modeStopOnFinalizedEpoch {
if q.highestExpectedSlot < q.blocksFetcher.bestFinalizedSlot() {
q.highestExpectedSlot = q.blocksFetcher.bestFinalizedSlot()
}
} else {
if q.highestExpectedSlot < q.blocksFetcher.bestNonFinalizedSlot() {
q.highestExpectedSlot = q.blocksFetcher.bestNonFinalizedSlot()
}
}
if nonSkippedSlot > q.highestExpectedSlot {
nonSkippedSlot = startSlot.Add(blocksPerRequest * (lookaheadSteps - 1))
}
q.smm.addStateMachine(nonSkippedSlot)
return nil
}