From 851a441a611c8d2a168db7fad82aac6cdd4c550c Mon Sep 17 00:00:00 2001 From: Alexey Sharp Date: Tue, 23 Mar 2021 15:29:34 +0000 Subject: [PATCH 1/3] Use proper forkID --- cmd/headers/download/sentry.go | 4 ++-- core/forkid/forkid.go | 21 +++++++-------------- 2 files changed, 9 insertions(+), 16 deletions(-) diff --git a/cmd/headers/download/sentry.go b/cmd/headers/download/sentry.go index 0728ffde7f3..f845f23d0f6 100644 --- a/cmd/headers/download/sentry.go +++ b/cmd/headers/download/sentry.go @@ -191,9 +191,9 @@ func runPeer( TD: gointerfaces.ConvertH256ToUint256Int(protoStatusData.TotalDifficulty).ToBig(), Head: gointerfaces.ConvertH256ToHash(protoStatusData.BestHash), Genesis: genesisHash, - ForkID: forkid.NewIDFromForks(protoStatusData.ForkData.Forks, genesisHash), + ForkID: forkid.NewIDFromForks(protoStatusData.ForkData.Forks, genesisHash, protoStatusData.MaxBlock), } - forkFilter := forkid.NewFilterFromForks(protoStatusData.ForkData.Forks, genesisHash) + forkFilter := forkid.NewFilterFromForks(protoStatusData.ForkData.Forks, genesisHash, protoStatusData.MaxBlock) networkID := protoStatusData.NetworkId if err := p2p.Send(rw, eth.StatusMsg, statusData); err != nil { return fmt.Errorf("handshake to peer %s: %v", peerID, err) diff --git a/core/forkid/forkid.go b/core/forkid/forkid.go index 6690e720f7d..685c03d67e1 100644 --- a/core/forkid/forkid.go +++ b/core/forkid/forkid.go @@ -55,12 +55,16 @@ type Filter func(id ID) error // NewID calculates the Ethereum fork ID from the chain config, genesis hash, and head. func NewID(config *params.ChainConfig, genesis common.Hash, head uint64) ID { + return NewIDFromForks(GatherForks(config), genesis, head) +} + +func NewIDFromForks(forks []uint64, genesis common.Hash, head uint64) ID { // Calculate the starting checksum from the genesis hash hash := crc32.ChecksumIEEE(genesis[:]) // Calculate the current fork checksum and the next fork block var next uint64 - for _, fork := range GatherForks(config) { + for _, fork := range forks { if fork <= head { // Fork already passed, checksum the previous hash and the fork number hash = checksumUpdate(hash, fork) @@ -72,17 +76,6 @@ func NewID(config *params.ChainConfig, genesis common.Hash, head uint64) ID { return ID{Hash: checksumToBytes(hash), Next: next} } -func NewIDFromForks(forks []uint64, genesis common.Hash) ID { - // Calculate the starting checksum from the genesis hash - hash := crc32.ChecksumIEEE(genesis[:]) - - // Calculate the current fork checksum and the next fork block - for _, fork := range forks { - hash = checksumUpdate(hash, fork) - } - return ID{Hash: checksumToBytes(hash), Next: 0} -} - // NewFilter creates a filter that returns if a fork ID should be rejected or notI // based on the local chain's status. func NewFilter(config *params.ChainConfig, genesis common.Hash, head func() uint64) Filter { @@ -94,8 +87,8 @@ func NewFilter(config *params.ChainConfig, genesis common.Hash, head func() uint ) } -func NewFilterFromForks(forks []uint64, genesis common.Hash) Filter { - head := func() uint64 { return 0 } +func NewFilterFromForks(forks []uint64, genesis common.Hash, headNumber uint64) Filter { + head := func() uint64 { return headNumber } return newFilter(forks, genesis, head) } From f9e3c4e83bfcd2d63fab88d40bbf87fc151c453f Mon Sep 17 00:00:00 2001 From: Alexey Sharp Date: Tue, 23 Mar 2021 17:16:00 +0000 Subject: [PATCH 2/3] Remove NaN from Write block bodies messages --- cmd/headers/download/downloader.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/cmd/headers/download/downloader.go b/cmd/headers/download/downloader.go index e0b717507a7..0e8ef918a52 100644 --- a/cmd/headers/download/downloader.go +++ b/cmd/headers/download/downloader.go @@ -503,8 +503,11 @@ func (cs *ControlServerImpl) blockBodies(inreq *proto_sentry.InboundMessage) err return fmt.Errorf("decode BlockBodies: %v", err) } delivered, undelivered := cs.bd.DeliverBodies(request) - // Approximate numbers - cs.bd.DeliverySize(float64(len(inreq.Data))*float64(delivered)/float64(delivered+undelivered), float64(len(inreq.Data))*float64(undelivered)/float64(delivered+undelivered)) + total := delivered + undelivered + if total > 0 { + // Approximate numbers + cs.bd.DeliverySize(float64(len(inreq.Data))*float64(delivered)/float64(delivered+undelivered), float64(len(inreq.Data))*float64(undelivered)/float64(delivered+undelivered)) + } return nil } From 44f382ae6336416f6b161acb4701add0042588bb Mon Sep 17 00:00:00 2001 From: Alexey Sharp Date: Tue, 23 Mar 2021 17:42:04 +0000 Subject: [PATCH 3/3] Move Prefetched blocks --- cmd/headers/download/downloader.go | 1 + eth/downloader/downloader_stagedsync.go | 3 +- eth/stagedsync/stage_bodies.go | 3 +- eth/stagedsync/stagebuilder.go | 3 +- eth/stagedsync/stagedsync.go | 5 ++-- eth/stagedsync/types.go | 4 ++- turbo/stages/bodydownload/body_algos.go | 29 +++++++++++++++---- turbo/stages/bodydownload/body_data_struct.go | 2 ++ .../stages/bodydownload}/prefetched_blocks.go | 2 +- 9 files changed, 39 insertions(+), 13 deletions(-) rename {eth/stagedsync => turbo/stages/bodydownload}/prefetched_blocks.go (97%) diff --git a/cmd/headers/download/downloader.go b/cmd/headers/download/downloader.go index 0e8ef918a52..55e89978622 100644 --- a/cmd/headers/download/downloader.go +++ b/cmd/headers/download/downloader.go @@ -484,6 +484,7 @@ func (cs *ControlServerImpl) newBlock(ctx context.Context, inreq *proto_sentry.I } else { return fmt.Errorf("singleHeaderAsSegment failed: %v", err) } + cs.bd.AddToPrefetch(request.Block) outreq := proto_sentry.PeerMinBlockRequest{ PeerId: inreq.PeerId, MinBlock: request.Block.NumberU64(), diff --git a/eth/downloader/downloader_stagedsync.go b/eth/downloader/downloader_stagedsync.go index d09086bb532..1e144d48ad3 100644 --- a/eth/downloader/downloader_stagedsync.go +++ b/eth/downloader/downloader_stagedsync.go @@ -11,6 +11,7 @@ import ( "github.com/ledgerwatch/turbo-geth/eth/stagedsync" "github.com/ledgerwatch/turbo-geth/log" "github.com/ledgerwatch/turbo-geth/rlp" + "github.com/ledgerwatch/turbo-geth/turbo/stages/bodydownload" ) // externsions for downloader needed for staged sync @@ -19,7 +20,7 @@ func (d *Downloader) SpawnBodyDownloadStage( id string, s *stagedsync.StageState, u stagedsync.Unwinder, - prefetchedBlocks *stagedsync.PrefetchedBlocks, + prefetchedBlocks *bodydownload.PrefetchedBlocks, ) (bool, error) { d.bodiesState = s d.bodiesUnwinder = u diff --git a/eth/stagedsync/stage_bodies.go b/eth/stagedsync/stage_bodies.go index 6bb6060c9c3..691902f5bfa 100644 --- a/eth/stagedsync/stage_bodies.go +++ b/eth/stagedsync/stage_bodies.go @@ -4,9 +4,10 @@ import ( "fmt" "github.com/ledgerwatch/turbo-geth/ethdb" + "github.com/ledgerwatch/turbo-geth/turbo/stages/bodydownload" ) -func spawnBodyDownloadStage(s *StageState, u Unwinder, d DownloaderGlue, pid string, pb *PrefetchedBlocks) error { +func spawnBodyDownloadStage(s *StageState, u Unwinder, d DownloaderGlue, pid string, pb *bodydownload.PrefetchedBlocks) error { logPrefix := s.state.LogPrefix() cont, err := d.SpawnBodyDownloadStage(logPrefix, pid, s, u, pb) if err != nil { diff --git a/eth/stagedsync/stagebuilder.go b/eth/stagedsync/stagebuilder.go index e0effe849e8..a8debf45815 100644 --- a/eth/stagedsync/stagebuilder.go +++ b/eth/stagedsync/stagebuilder.go @@ -17,6 +17,7 @@ import ( "github.com/ledgerwatch/turbo-geth/log" "github.com/ledgerwatch/turbo-geth/params" "github.com/ledgerwatch/turbo-geth/turbo/shards" + "github.com/ledgerwatch/turbo-geth/turbo/stages/bodydownload" ) type ChainEventNotifier interface { @@ -45,7 +46,7 @@ type StageParameters struct { headersFetchers []func() error txPool *core.TxPool poolStart func() error - prefetchedBlocks *PrefetchedBlocks + prefetchedBlocks *bodydownload.PrefetchedBlocks stateReaderBuilder StateReaderBuilder stateWriterBuilder StateWriterBuilder notifier ChainEventNotifier diff --git a/eth/stagedsync/stagedsync.go b/eth/stagedsync/stagedsync.go index 7e7a891c800..506b3a286f3 100644 --- a/eth/stagedsync/stagedsync.go +++ b/eth/stagedsync/stagedsync.go @@ -9,12 +9,13 @@ import ( "github.com/ledgerwatch/turbo-geth/ethdb" "github.com/ledgerwatch/turbo-geth/params" "github.com/ledgerwatch/turbo-geth/turbo/shards" + "github.com/ledgerwatch/turbo-geth/turbo/stages/bodydownload" ) const prof = false // whether to profile type StagedSync struct { - PrefetchedBlocks *PrefetchedBlocks + PrefetchedBlocks *bodydownload.PrefetchedBlocks stageBuilders StageBuilders unwindOrder UnwindOrder params OptionalParameters @@ -40,7 +41,7 @@ type OptionalParameters struct { func New(stages StageBuilders, unwindOrder UnwindOrder, params OptionalParameters) *StagedSync { return &StagedSync{ - PrefetchedBlocks: NewPrefetchedBlocks(), + PrefetchedBlocks: bodydownload.NewPrefetchedBlocks(), stageBuilders: stages, unwindOrder: unwindOrder, params: params, diff --git a/eth/stagedsync/types.go b/eth/stagedsync/types.go index 60b05747e93..a06d0fb52cb 100644 --- a/eth/stagedsync/types.go +++ b/eth/stagedsync/types.go @@ -1,6 +1,8 @@ package stagedsync +import "github.com/ledgerwatch/turbo-geth/turbo/stages/bodydownload" + type DownloaderGlue interface { SpawnHeaderDownloadStage([]func() error, *StageState, Unwinder) error - SpawnBodyDownloadStage(string, string, *StageState, Unwinder, *PrefetchedBlocks) (bool, error) + SpawnBodyDownloadStage(string, string, *StageState, Unwinder, *bodydownload.PrefetchedBlocks) (bool, error) } diff --git a/turbo/stages/bodydownload/body_algos.go b/turbo/stages/bodydownload/body_algos.go index 6a63271bf5f..c3ea23b74bb 100644 --- a/turbo/stages/bodydownload/body_algos.go +++ b/turbo/stages/bodydownload/body_algos.go @@ -102,12 +102,17 @@ func (bd *BodyDownload) RequestMoreBodies(db ethdb.Database, blockNum uint64, cu log.Error("Could not find canonical header", "block number", blockNum) } if header != nil { - bd.deliveries[blockNum-bd.requestedLow] = types.NewBlockWithHeader(header) // Block without uncles and transactions - if header.UncleHash != types.EmptyUncleHash || header.TxHash != types.EmptyRootHash { - var doubleHash DoubleHash - copy(doubleHash[:], header.UncleHash.Bytes()) - copy(doubleHash[common.HashLength:], header.TxHash.Bytes()) - bd.requestedMap[doubleHash] = blockNum + if block := bd.prefetchedBlocks.Pop(hash); block != nil { + // Block is prefetched, no need to request + bd.deliveries[blockNum-bd.requestedLow] = block + } else { + bd.deliveries[blockNum-bd.requestedLow] = types.NewBlockWithHeader(header) // Block without uncles and transactions + if header.UncleHash != types.EmptyUncleHash || header.TxHash != types.EmptyRootHash { + var doubleHash DoubleHash + copy(doubleHash[:], header.UncleHash.Bytes()) + copy(doubleHash[common.HashLength:], header.TxHash.Bytes()) + bd.requestedMap[doubleHash] = blockNum + } } } } @@ -243,3 +248,15 @@ func (bd *BodyDownload) PrintPeerMap() { fmt.Printf("---------------------------\n") bd.peerMap = make(map[string]int) } + +func (bd *BodyDownload) AddToPrefetch(block *types.Block) { + if hash := types.CalcUncleHash(block.Uncles()); hash != block.UncleHash() { + log.Warn("Propagated block has invalid uncles", "have", hash, "exp", block.UncleHash()) + return + } + if hash := types.DeriveSha(block.Transactions()); hash != block.TxHash() { + log.Warn("Propagated block has invalid body", "have", hash, "exp", block.TxHash()) + return + } + bd.prefetchedBlocks.Add(block) +} diff --git a/turbo/stages/bodydownload/body_data_struct.go b/turbo/stages/bodydownload/body_data_struct.go index 08cc5682fd6..4a41745e92f 100644 --- a/turbo/stages/bodydownload/body_data_struct.go +++ b/turbo/stages/bodydownload/body_data_struct.go @@ -28,6 +28,7 @@ type BodyDownload struct { lowWaitUntil uint64 // Time to wait for before starting the next round request from requestedLow outstandingLimit uint64 // Limit of number of outstanding blocks for body requests peerMap map[string]int + prefetchedBlocks *PrefetchedBlocks } // BodyRequest is a sketch of the request for block bodies, meaning that access to the database is required to convert it to the actual BlockBodies request (look up hashes of canonical blocks) @@ -47,6 +48,7 @@ func NewBodyDownload(outstandingLimit int) *BodyDownload { deliveries: make([]*types.Block, outstandingLimit+MaxBodiesInRequest), requests: make([]*BodyRequest, outstandingLimit+MaxBodiesInRequest), peerMap: make(map[string]int), + prefetchedBlocks: NewPrefetchedBlocks(), } return bd } diff --git a/eth/stagedsync/prefetched_blocks.go b/turbo/stages/bodydownload/prefetched_blocks.go similarity index 97% rename from eth/stagedsync/prefetched_blocks.go rename to turbo/stages/bodydownload/prefetched_blocks.go index 5cd2aa0f3cf..4f95b2e201a 100644 --- a/eth/stagedsync/prefetched_blocks.go +++ b/turbo/stages/bodydownload/prefetched_blocks.go @@ -1,4 +1,4 @@ -package stagedsync +package bodydownload import ( "github.com/ledgerwatch/turbo-geth/common"