Correct forkID computation in sentry, use blocks prefetched by NewBlockPacket, Fix NaN delivery speed output

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(),
@@ -503,8 +504,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
 }
 

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)

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)
 }
 

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

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 {

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

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,

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)
 }

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)
+}

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
 }

eth/stagedsync/prefetched_blocks.go → turbo/stages/bodydownload/prefetched_blocks.go

@@ -1,4 +1,4 @@
-package stagedsync
+package bodydownload
 
 import (
 	"github.com/ledgerwatch/turbo-geth/common"