New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Batch save blocks for initial sync. 80% faster BPS #5215
Changes from 24 commits
512ef28
7a370b0
5ef3f0b
96d78ea
4197c1d
15d629e
cf92a43
b6013af
7e8c0b9
14e6379
7e7273a
9ba5b11
0281a5f
8df754c
1b6f64c
09729aa
fc2589e
1b7f614
3b52bd1
9e88161
280ea97
e7ee4d5
a0fad8a
271b489
add721e
acfba22
ef3263f
f3b415f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -5,9 +5,11 @@ import ( | |
"sort" | ||
|
||
"github.com/pkg/errors" | ||
ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1" | ||
"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers" | ||
stateTrie "github.com/prysmaticlabs/prysm/beacon-chain/state" | ||
"github.com/prysmaticlabs/prysm/shared/bytesutil" | ||
"github.com/prysmaticlabs/prysm/shared/featureconfig" | ||
"github.com/prysmaticlabs/prysm/shared/params" | ||
) | ||
|
||
|
@@ -170,10 +172,23 @@ func (s *Service) generateState(ctx context.Context, startRoot [32]byte, endRoot | |
if preState == nil { | ||
return nil, errors.New("finalized state does not exist in db") | ||
} | ||
endBlock, err := s.beaconDB.Block(ctx, endRoot) | ||
if err != nil { | ||
return nil, err | ||
|
||
var endBlock *ethpb.SignedBeaconBlock | ||
if featureconfig.Get().InitSyncBatchSaveBlocks && s.hasInitSyncBlock(endRoot) { | ||
endBlock = s.getInitSyncBlock(endRoot) | ||
if featureconfig.Get().InitSyncBatchSaveBlocks { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. why check again ? this will always be true There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. fixed! |
||
if err := s.beaconDB.SaveBlocks(ctx, s.getInitSyncBlocks()); err != nil { | ||
return nil, err | ||
} | ||
s.clearInitSyncBlocks() | ||
} | ||
} else { | ||
endBlock, err = s.beaconDB.Block(ctx, endRoot) | ||
if err != nil { | ||
return nil, err | ||
} | ||
} | ||
|
||
if endBlock == nil { | ||
return nil, errors.New("provided block root does not have block saved in the db") | ||
} | ||
|
@@ -189,3 +204,48 @@ func (s *Service) generateState(ctx context.Context, startRoot [32]byte, endRoot | |
} | ||
return postState, nil | ||
} | ||
|
||
// This saves a beacon block to the initial sync blocks cache. | ||
func (s *Service) saveInitSyncBlock(r [32]byte, b *ethpb.SignedBeaconBlock) { | ||
s.initSyncBlocksLock.Lock() | ||
defer s.initSyncBlocksLock.Unlock() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. do we need a lock for this ? initial sync is single-threaded , so there isnt a risk of concurrent writes There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I tried without the lock, the performance is neglectable. Just figure be safe than sorry later if initial sync moves to multi threaded |
||
s.initSyncBlocks[r] = b | ||
} | ||
|
||
// This checks if a beacon block exists in the initial sync blocks cache using the root | ||
// of the block. | ||
func (s *Service) hasInitSyncBlock(r [32]byte) bool { | ||
s.initSyncBlocksLock.RLock() | ||
defer s.initSyncBlocksLock.RUnlock() | ||
_, ok := s.initSyncBlocks[r] | ||
return ok | ||
} | ||
|
||
// This retrieves a beacon block from the initial sync blocks cache using the root of | ||
// the block. | ||
func (s *Service) getInitSyncBlock(r [32]byte) *ethpb.SignedBeaconBlock { | ||
s.initSyncBlocksLock.RLock() | ||
defer s.initSyncBlocksLock.RUnlock() | ||
b := s.initSyncBlocks[r] | ||
return b | ||
} | ||
|
||
// This retrieves all the beacon blocks from the initial sync blocks cache, the returned | ||
// blocks are unordered. | ||
func (s *Service) getInitSyncBlocks() []*ethpb.SignedBeaconBlock { | ||
s.initSyncBlocksLock.RLock() | ||
defer s.initSyncBlocksLock.RUnlock() | ||
|
||
blks := make([]*ethpb.SignedBeaconBlock, 0, len(s.initSyncBlocks)) | ||
for _, b := range s.initSyncBlocks { | ||
blks = append(blks, b) | ||
} | ||
return blks | ||
} | ||
|
||
// This clears out the initial sync blocks cache. | ||
func (s *Service) clearInitSyncBlocks() { | ||
s.initSyncBlocksLock.Lock() | ||
defer s.initSyncBlocksLock.Unlock() | ||
s.initSyncBlocks = make(map[[32]byte]*ethpb.SignedBeaconBlock) | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -15,6 +15,7 @@ import ( | |
"github.com/prysmaticlabs/prysm/shared/attestationutil" | ||
"github.com/prysmaticlabs/prysm/shared/bytesutil" | ||
"github.com/prysmaticlabs/prysm/shared/featureconfig" | ||
"github.com/prysmaticlabs/prysm/shared/params" | ||
"github.com/sirupsen/logrus" | ||
"go.opencensus.io/trace" | ||
) | ||
|
@@ -210,13 +211,17 @@ func (s *Service) onBlockInitialSyncStateTransition(ctx context.Context, signed | |
return errors.Wrap(err, "could not execute state transition") | ||
} | ||
|
||
if err := s.beaconDB.SaveBlock(ctx, signed); err != nil { | ||
return errors.Wrapf(err, "could not save block from slot %d", b.Slot) | ||
} | ||
root, err := stateutil.BlockRoot(b) | ||
if err != nil { | ||
return errors.Wrapf(err, "could not get signing root of block %d", b.Slot) | ||
} | ||
if featureconfig.Get().InitSyncBatchSaveBlocks { | ||
s.saveInitSyncBlock(root, signed) | ||
} else { | ||
if err := s.beaconDB.SaveBlock(ctx, signed); err != nil { | ||
return errors.Wrapf(err, "could not save block from slot %d", b.Slot) | ||
} | ||
} | ||
|
||
if err := s.insertBlockToForkChoiceStore(ctx, b, root, postState); err != nil { | ||
return errors.Wrapf(err, "could not insert block %d to fork choice store", b.Slot) | ||
|
@@ -247,6 +252,14 @@ func (s *Service) onBlockInitialSyncStateTransition(ctx context.Context, signed | |
} | ||
} | ||
|
||
// Rate limit how many blocks (2 epochs worth of blocks) a node keeps in the memory. | ||
if len(s.getInitSyncBlocks()) > 2*int(params.BeaconConfig().SlotsPerEpoch) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. can we make this value a configurable constant at the start of the file There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. yep, done! |
||
if err := s.beaconDB.SaveBlocks(ctx, s.getInitSyncBlocks()); err != nil { | ||
return err | ||
} | ||
s.clearInitSyncBlocks() | ||
} | ||
|
||
// Update finalized check point. Prune the block cache and helper caches on every new finalized epoch. | ||
if postState.FinalizedCheckpointEpoch() > s.finalizedCheckpt.Epoch { | ||
if !featureconfig.Get().NewStateMgmt { | ||
|
@@ -264,6 +277,13 @@ func (s *Service) onBlockInitialSyncStateTransition(ctx context.Context, signed | |
} | ||
} | ||
|
||
if featureconfig.Get().InitSyncBatchSaveBlocks { | ||
if err := s.beaconDB.SaveBlocks(ctx, s.getInitSyncBlocks()); err != nil { | ||
return err | ||
} | ||
s.clearInitSyncBlocks() | ||
} | ||
|
||
if err := s.beaconDB.SaveFinalizedCheckpoint(ctx, postState.FinalizedCheckpoint()); err != nil { | ||
return errors.Wrap(err, "could not save finalized checkpoint") | ||
} | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -11,7 +11,6 @@ import ( | |
|
||
"github.com/pkg/errors" | ||
ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1" | ||
"github.com/prysmaticlabs/go-ssz" | ||
"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers" | ||
"github.com/prysmaticlabs/prysm/beacon-chain/db/filters" | ||
"github.com/prysmaticlabs/prysm/shared/bytesutil" | ||
|
@@ -227,16 +226,16 @@ func (k *Store) SaveBlocks(ctx context.Context, blocks []*ethpb.SignedBeaconBloc | |
defer span.End() | ||
|
||
return k.db.Update(func(tx *bolt.Tx) error { | ||
bkt := tx.Bucket(blocksBucket) | ||
for _, block := range blocks { | ||
if err := k.setBlockSlotBitField(ctx, tx, block.Block.Slot); err != nil { | ||
return err | ||
} | ||
|
||
blockRoot, err := ssz.HashTreeRoot(block.Block) | ||
blockRoot, err := stateutil.BlockRoot(block.Block) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Nice |
||
if err != nil { | ||
return err | ||
} | ||
bkt := tx.Bucket(blocksBucket) | ||
|
||
if existingBlock := bkt.Get(blockRoot[:]); existingBlock != nil { | ||
continue | ||
} | ||
|
@@ -249,6 +248,7 @@ func (k *Store) SaveBlocks(ctx context.Context, blocks []*ethpb.SignedBeaconBloc | |
return errors.Wrap(err, "could not update DB indices") | ||
} | ||
k.blockCache.Set(string(blockRoot[:]), block, int64(len(enc))) | ||
|
||
if err := bkt.Put(blockRoot[:], enc); err != nil { | ||
return err | ||
} | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -181,7 +181,17 @@ func (s *Service) roundRobinSync(genesis time.Time) error { | |
} | ||
} | ||
} | ||
startBlock := s.chain.HeadSlot() + 1 | ||
var startBlock uint64 | ||
if featureconfig.Get().InitSyncBatchSaveBlocks { | ||
lastFinalizedEpoch := s.chain.FinalizedCheckpt().Epoch | ||
lastFinalizedState, err := s.db.HighestSlotStatesBelow(ctx, helpers.StartSlot(lastFinalizedEpoch)) | ||
if err != nil { | ||
return err | ||
} | ||
startBlock = lastFinalizedState[0].Slot() + 1 | ||
} else { | ||
startBlock = s.chain.HeadSlot() + 1 | ||
} | ||
skippedBlocks := blockBatchSize * uint64(lastEmptyRequests*len(peers)) | ||
if startBlock+skippedBlocks > helpers.StartSlot(finalizedEpoch+1) { | ||
log.WithField("finalizedEpoch", finalizedEpoch).Debug("Requested block range is greater than the finalized epoch") | ||
|
@@ -193,7 +203,7 @@ func (s *Service) roundRobinSync(genesis time.Time) error { | |
1, // step | ||
blockBatchSize, // count | ||
peers, // peers | ||
0, // remainder | ||
0, // reminder | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Reminder? What does that mean in this context? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Sorry it should have been remainder |
||
) | ||
if err != nil { | ||
log.WithError(err).Error("Round robing sync request failed") | ||
|
@@ -209,10 +219,12 @@ func (s *Service) roundRobinSync(genesis time.Time) error { | |
|
||
for _, blk := range blocks { | ||
s.logSyncStatus(genesis, blk.Block, peers, counter) | ||
if !s.db.HasBlock(ctx, bytesutil.ToBytes32(blk.Block.ParentRoot)) { | ||
log.Debugf("Beacon node doesn't have a block in db with root %#x", blk.Block.ParentRoot) | ||
parentRoot := bytesutil.ToBytes32(blk.Block.ParentRoot) | ||
if !s.db.HasBlock(ctx, parentRoot) && !s.chain.HasInitSyncBlock(parentRoot) { | ||
log.Debugf("Beacon node doesn't have a block in db or cache with root %#x", parentRoot) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Should this be changed to There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Also, not feature flag needed here? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Updated to WithField(), feature flag not needed in this csae |
||
continue | ||
} | ||
|
||
s.blockNotifier.BlockFeed().Send(&feed.Event{ | ||
Type: blockfeed.ReceivedBlock, | ||
Data: &blockfeed.ReceivedBlockData{SignedBlock: blk}, | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why is this condition here? Wouldn't it already be true?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Oops, thanks!