github/prysm
1
1
Fork 0
mirror of https://github.com/OffchainLabs/prysm.git synced 2026-01-09 15:37:56 -05:00
Code Issues Packages Projects Releases Wiki Activity

Do Not Subscribe to Blocks in Initial Sync (#2524)

Browse Source 
* only sub to block batches

* batch sub remove

* tests

* fix lint

* gazelle

* delete old im mem blocks code
This commit is contained in:
Raul Jordan
2019-05-07 21:12:36 -05:00
committed by GitHub
parent 7642f950d8
commit 76881fd1ae
4 changed files with 11 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
beacon-chain/sync/initial-sync/BUILD.bazel
View File

@@ -45,7 +45,6 @@ go_test(
"//shared/p2p:go_default_library",
"//shared/params:go_default_library",
"//shared/testutil:go_default_library",
"@com_github_ethereum_go_ethereum//common:go_default_library",
"@com_github_gogo_protobuf//proto:go_default_library",
"@com_github_libp2p_go_libp2p_peer//:go_default_library",
"@com_github_sirupsen_logrus//hooks/test:go_default_library",

64
beacon-chain/sync/initial-sync/service.go
View File

@@ -36,16 +36,14 @@ var debugError = "debug:"
// Config defines the configurable properties of InitialSync.
//
type Config struct {
SyncPollingInterval time.Duration
BlockBufferSize int
BlockAnnounceBufferSize int
BatchedBlockBufferSize int
StateBufferSize int
BeaconDB *db.BeaconDB
P2P p2pAPI
SyncService syncService
ChainService chainService
PowChain powChainService
SyncPollingInterval time.Duration
BatchedBlockBufferSize int
StateBufferSize int
BeaconDB *db.BeaconDB
P2P p2pAPI
SyncService syncService
ChainService chainService
PowChain powChainService
}
// DefaultConfig provides the default configuration for a sync service.
@@ -54,11 +52,9 @@ type Config struct {
// StateBufferSize determines the buffer size of the `stateBuf` channel.
func DefaultConfig() *Config {
return &Config{
SyncPollingInterval: time.Duration(params.BeaconConfig().SyncPollingInterval) * time.Second,
BlockBufferSize: params.BeaconConfig().DefaultBufferSize,
BatchedBlockBufferSize: params.BeaconConfig().DefaultBufferSize,
BlockAnnounceBufferSize: params.BeaconConfig().DefaultBufferSize,
StateBufferSize: params.BeaconConfig().DefaultBufferSize,
SyncPollingInterval: time.Duration(params.BeaconConfig().SyncPollingInterval) * time.Second,
BatchedBlockBufferSize: params.BeaconConfig().DefaultBufferSize,
StateBufferSize: params.BeaconConfig().DefaultBufferSize,
}
}
@@ -93,16 +89,13 @@ type InitialSync struct {
chainService chainService
db *db.BeaconDB
powchain powChainService
blockAnnounceBuf chan p2p.Message
batchedBlockBuf chan p2p.Message
blockBuf chan p2p.Message
stateBuf chan p2p.Message
currentSlot uint64
highestObservedSlot uint64
highestObservedRoot [32]byte
beaconStateSlot uint64
syncPollingInterval time.Duration
inMemoryBlocks map[uint64]*pb.BeaconBlock
syncedFeed *event.Feed
stateReceived bool
lastRequestedSlot uint64
@@ -119,9 +112,7 @@ func NewInitialSyncService(ctx context.Context,
) *InitialSync {
ctx, cancel := context.WithCancel(ctx)
blockBuf := make(chan p2p.Message, cfg.BlockBufferSize)
stateBuf := make(chan p2p.Message, cfg.StateBufferSize)
blockAnnounceBuf := make(chan p2p.Message, cfg.BlockAnnounceBufferSize)
batchedBlockBuf := make(chan p2p.Message, cfg.BatchedBlockBufferSize)
return &InitialSync{
@@ -135,12 +126,9 @@ func NewInitialSyncService(ctx context.Context,
currentSlot: params.BeaconConfig().GenesisSlot,
highestObservedSlot: params.BeaconConfig().GenesisSlot,
beaconStateSlot: params.BeaconConfig().GenesisSlot,
blockBuf: blockBuf,
stateBuf: stateBuf,
batchedBlockBuf: batchedBlockBuf,
blockAnnounceBuf: blockAnnounceBuf,
syncPollingInterval: cfg.SyncPollingInterval,
inMemoryBlocks: map[uint64]*pb.BeaconBlock{},
syncedFeed: new(event.Feed),
stateReceived: false,
mutex: new(sync.Mutex),
@@ -155,7 +143,6 @@ func (s *InitialSync) Start() {
}
s.currentSlot = cHead.Slot
go s.run()
go s.checkInMemoryBlocks()
}
// Stop kills the initial sync goroutine.
@@ -259,40 +246,16 @@ func (s *InitialSync) exitInitialSync(ctx context.Context, block *pb.BeaconBlock
return nil
}
// checkInMemoryBlocks is another routine which will run concurrently with the
// main routine for initial sync, where it checks the blocks saved in memory regularly
// to see if the blocks are valid enough to be processed.
func (s *InitialSync) checkInMemoryBlocks() {
for {
select {
case <-s.ctx.Done():
return
default:
if s.currentSlot == s.highestObservedSlot {
return
}
s.mutex.Lock()
if block, ok := s.inMemoryBlocks[s.currentSlot+1]; ok && s.currentSlot+1 <= s.highestObservedSlot {
s.processBlock(s.ctx, block)
}
s.mutex.Unlock()
}
}
}
// run is the main goroutine for the initial sync service.
// delayChan is explicitly passed into this function to facilitate tests that don't require a timeout.
// It is assumed that the goroutine `run` is only called once per instance.
func (s *InitialSync) run() {
blockSub := s.p2p.Subscribe(&pb.BeaconBlockResponse{}, s.blockBuf)
batchedBlocksub := s.p2p.Subscribe(&pb.BatchedBeaconBlockResponse{}, s.batchedBlockBuf)
beaconStateSub := s.p2p.Subscribe(&pb.BeaconStateResponse{}, s.stateBuf)
defer func() {
blockSub.Unsubscribe()
beaconStateSub.Unsubscribe()
batchedBlocksub.Unsubscribe()
close(s.batchedBlockBuf)
close(s.blockBuf)
close(s.stateBuf)
}()
@@ -306,11 +269,6 @@ func (s *InitialSync) run() {
case <-s.ctx.Done():
log.Debug("Exiting goroutine")
return
case msg := <-s.blockBuf:
safelyHandleMessage(func(message p2p.Message) {
data := message.Data.(*pb.BeaconBlockResponse)
s.processBlock(message.Ctx, data.Block)
}, msg)
case msg := <-s.stateBuf:
safelyHandleMessage(s.processState, msg)
case msg := <-s.batchedBlockBuf:

127
beacon-chain/sync/initial-sync/service_test.go
View File

@@ -2,11 +2,9 @@ package initialsync
import (
"context"
"math/big"
"testing"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/gogo/protobuf/proto"
peer "github.com/libp2p/go-libp2p-peer"
b "github.com/prysmaticlabs/prysm/beacon-chain/core/blocks"
@@ -51,12 +49,6 @@ func (ms *mockSyncService) ResumeSync() {
}
type mockPowchain struct{}
func (mp *mockPowchain) BlockExists(ctx context.Context, hash common.Hash) (bool, *big.Int, error) {
return true, nil, nil
}
type mockChainService struct{}
func (ms *mockChainService) CanonicalBlockFeed() *event.Feed {
@@ -114,125 +106,6 @@ func setUpGenesisStateAndBlock(beaconDB *db.BeaconDB, t *testing.T) {
}
}
func TestSavingBlock_InSync(t *testing.T) {
hook := logTest.NewGlobal()
db := internal.SetupDB(t)
setUpGenesisStateAndBlock(db, t)
cfg := &Config{
P2P: &mockP2P{},
SyncService: &mockSyncService{},
ChainService: &mockChainService{},
BeaconDB: db,
PowChain: &mockPowchain{},
}
ss := NewInitialSyncService(context.Background(), cfg)
exitRoutine := make(chan bool)
defer func() {
close(exitRoutine)
}()
go func() {
ss.run()
exitRoutine <- true
}()
genericHash := make([]byte, 32)
genericHash[0] = 'a'
fState := &pb.BeaconState{
FinalizedEpoch: params.BeaconConfig().GenesisEpoch + 1,
LatestBlock: &pb.BeaconBlock{
Slot: params.BeaconConfig().GenesisSlot + params.BeaconConfig().SlotsPerEpoch,
},
LatestEth1Data: &pb.Eth1Data{
BlockHash32: []byte{},
},
}
stateResponse := &pb.BeaconStateResponse{
FinalizedState: fState,
}
incorrectState := &pb.BeaconState{
FinalizedEpoch: params.BeaconConfig().GenesisEpoch,
JustifiedEpoch: params.BeaconConfig().GenesisEpoch + 1,
LatestBlock: &pb.BeaconBlock{
Slot: params.BeaconConfig().GenesisSlot + 4*params.BeaconConfig().SlotsPerEpoch,
},
LatestEth1Data: &pb.Eth1Data{
BlockHash32: []byte{},
},
}
incorrectStateResponse := &pb.BeaconStateResponse{
FinalizedState: incorrectState,
}
stateRoot, err := hashutil.HashProto(fState)
if err != nil {
t.Fatalf("unable to tree hash state: %v", err)
}
beaconStateRootHash32 := stateRoot
getBlockResponseMsg := func(Slot uint64) p2p.Message {
block := &pb.BeaconBlock{
Eth1Data: &pb.Eth1Data{
DepositRootHash32: []byte{1, 2, 3},
BlockHash32: []byte{4, 5, 6},
},
ParentRootHash32: genericHash,
Slot: Slot,
StateRootHash32: beaconStateRootHash32[:],
}
blockResponse := &pb.BeaconBlockResponse{
Block: block,
}
return p2p.Message{
Peer: "",
Data: blockResponse,
Ctx: context.Background(),
}
}
if err != nil {
t.Fatalf("Unable to hash block %v", err)
}
msg1 := getBlockResponseMsg(params.BeaconConfig().GenesisSlot + 1)
// saving genesis block
ss.blockBuf <- msg1
msg2 := p2p.Message{
Peer: "",
Data: incorrectStateResponse,
Ctx: context.Background(),
}
ss.stateBuf <- msg2
msg2.Data = stateResponse
ss.stateBuf <- msg2
msg1 = getBlockResponseMsg(params.BeaconConfig().GenesisSlot + 1)
ss.blockBuf <- msg1
msg1 = getBlockResponseMsg(params.BeaconConfig().GenesisSlot + 2)
ss.blockBuf <- msg1
ss.cancel()
<-exitRoutine
hook.Reset()
internal.TeardownDB(t, db)
}
func TestProcessingBatchedBlocks_OK(t *testing.T) {
db := internal.SetupDB(t)
defer internal.TeardownDB(t, db)

27
beacon-chain/sync/initial-sync/sync_blocks.go
View File

@@ -36,29 +36,6 @@ func (s *InitialSync) processBlock(ctx context.Context, block *pb.BeaconBlock) {
return
}
// if it isn't the block in the next slot we check if it is a skipped slot.
// if it isn't skipped we save it in memory.
if block.Slot != (s.currentSlot + 1) {
// if parent exists we validate the block.
if s.doesParentExist(block) {
if err := s.validateAndSaveNextBlock(ctx, block); err != nil {
// Debug error so as not to have noisy error logs
if strings.HasPrefix(err.Error(), debugError) {
log.Debug(strings.TrimPrefix(err.Error(), debugError))
return
}
log.Errorf("Unable to save block: %v", err)
}
return
}
s.mutex.Lock()
defer s.mutex.Unlock()
if _, ok := s.inMemoryBlocks[block.Slot]; !ok {
s.inMemoryBlocks[block.Slot] = block
}
return
}
if err := s.validateAndSaveNextBlock(ctx, block); err != nil {
// Debug error so as not to have noisy error logs
if strings.HasPrefix(err.Error(), debugError) {
@@ -148,10 +125,6 @@ func (s *InitialSync) validateAndSaveNextBlock(ctx context.Context, block *pb.Be
s.mutex.Lock()
defer s.mutex.Unlock()
// delete block from memory.
if _, ok := s.inMemoryBlocks[block.Slot]; ok {
delete(s.inMemoryBlocks, block.Slot)
}
state, err := s.db.HeadState(ctx)
if err != nil {
return err