BlockFetcher.loop¶
| Source (upstream pre-purge) | Current | |
|---|---|---|
| File | eth/fetcher/block_fetcher.go |
block_fetcher_pow.go |
| Symbol | BlockFetcher.loop |
BlockFetcher.loop |
| Ref | f4d53133f~1 |
etc/v1.17.3-full-node |
WithBody API changed upstream in #29482 — now takes types.Body{} instead of separate args
3-way merge — purge → getc ← upstream¶
pre-purge≈ adapted (origin inferred by similarity)
↗// Loop is the main fetcher loop, checking and processing various notification
// events.
func (f *BlockFetcher) loop() {
// Iterate the block fetching until a quit is requested
var (
fetchTimer = time.NewTimer(0)
completeTimer = time.NewTimer(0)
)
<-fetchTimer.C // clear out the channel
<-completeTimer.C
defer fetchTimer.Stop()
defer completeTimer.Stop()
↗ for {
// Clean up any expired block fetches
for hash, announce := range f.fetching {
if time.Since(announce.time) > fetchTimeout {
f.forgetHash(hash)
}
}
// Import any queued blocks that could potentially fit
height := f.chainHeight()
for !f.queue.Empty() {
op := f.queue.PopItem()
hash := op.hash()
if f.queueChangeHook != nil {
f.queueChangeHook(hash, false)
}
// If too high up the chain or phase, continue later
number := op.number()
if number > height+1 {
f.queue.Push(op, -int64(number))
if f.queueChangeHook != nil {
f.queueChangeHook(hash, true)
}
break
}
// Otherwise if fresh and still unknown, try and import
if (number+maxUncleDist < height) || (f.light && f.getHeader(hash) != nil) || (!f.light && f.getBlock(hash) != nil) {
f.forgetBlock(hash)
continue
}
if f.light {
f.importHeaders(op.origin, op.header)
} else {
f.importBlocks(op.origin, op.block)
}
}
// Wait for an outside event to occur
select {
case <-f.quit:
// BlockFetcher terminating, abort all operations
return
↗ case notification := <-f.notify:
// A block was announced, make sure the peer isn't DOSing us
blockAnnounceInMeter.Mark(1)
↗ count := f.announces[notification.origin] + 1
if count > hashLimit {
log.Debug("Peer exceeded outstanding announces", "peer", notification.origin, "limit", hashLimit)
blockAnnounceDOSMeter.Mark(1)
break
}
if notification.number == 0 {
break
}
// If we have a valid block number, check that it's potentially useful
if dist := int64(notification.number) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist {
log.Debug("Peer discarded announcement", "peer", notification.origin, "number", notification.number, "hash", notification.hash, "distance", dist)
blockAnnounceDropMeter.Mark(1)
break
}
// All is well, schedule the announce if block's not yet downloading
if _, ok := f.fetching[notification.hash]; ok {
break
}
if _, ok := f.completing[notification.hash]; ok {
break
}
f.announces[notification.origin] = count
f.announced[notification.hash] = append(f.announced[notification.hash], notification)
if f.announceChangeHook != nil && len(f.announced[notification.hash]) == 1 {
f.announceChangeHook(notification.hash, true)
}
if len(f.announced) == 1 {
f.rescheduleFetch(fetchTimer)
}
↗ case op := <-f.inject:
// A direct block insertion was requested, try and fill any pending gaps
blockBroadcastInMeter.Mark(1)
↗ // Now only direct block injection is allowed, drop the header injection
// here silently if we receive.
if f.light {
continue
}
f.enqueue(op.origin, nil, op.block)
↗ case hash := <-f.done:
// A pending import finished, remove all traces of the notification
f.forgetHash(hash)
f.forgetBlock(hash)
↗ case <-fetchTimer.C:
// At least one block's timer ran out, check for needing retrieval
request := make(map[string][]common.Hash)
↗ for hash, announces := range f.announced {
// In current LES protocol(les2/les3), only header announce is
// available, no need to wait too much time for header broadcast.
timeout := arriveTimeout - gatherSlack
if f.light {
timeout = 0
}
if time.Since(announces[0].time) > timeout {
// Pick a random peer to retrieve from, reset all others
announce := announces[rand.Intn(len(announces))]
f.forgetHash(hash)
↗ // If the block still didn't arrive, queue for fetching
if (f.light && f.getHeader(hash) == nil) || (!f.light && f.getBlock(hash) == nil) {
request[announce.origin] = append(request[announce.origin], hash)
f.fetching[hash] = announce
}
}
}
// Send out all block header requests
for peer, hashes := range request {
log.Trace("Fetching scheduled headers", "peer", peer, "list", hashes)
↗ // Create a closure of the fetch and schedule in on a new thread
≈ fetchHeader := f.fetching[hashes[0]].fetchHeader
↗ go func(peer string) {
if f.fetchingHook != nil {
f.fetchingHook(hashes)
}
for _, hash := range hashes {
headerFetchMeter.Mark(1)
go func(hash common.Hash) {
resCh := make(chan *eth.Response)
↗ req, err := fetchHeader(hash, resCh)
if err != nil {
return // Legacy code, yolo
}
defer req.Close()
↗ timeout := time.NewTimer(2 * fetchTimeout) // 2x leeway before dropping the peer
defer timeout.Stop()
↗ select {
case res := <-resCh:
res.Done <- nil
f.FilterHeaders(peer, *res.Res.(*eth.BlockHeadersRequest), time.Now())
↗ case <-timeout.C:
// The peer didn't respond in time. The request
// was already rescheduled at this point, we were
// waiting for a catchup. With an unresponsive
// peer however, it's a protocol violation.
f.dropPeer(peer)
}
}(hash)
}
}(peer)
}
// Schedule the next fetch if blocks are still pending
f.rescheduleFetch(fetchTimer)
↗ case <-completeTimer.C:
// At least one header's timer ran out, retrieve everything
request := make(map[string][]common.Hash)
↗ for hash, announces := range f.fetched {
// Pick a random peer to retrieve from, reset all others
announce := announces[rand.Intn(len(announces))]
f.forgetHash(hash)
↗ // If the block still didn't arrive, queue for completion
if f.getBlock(hash) == nil {
request[announce.origin] = append(request[announce.origin], hash)
f.completing[hash] = announce
}
}
// Send out all block body requests
for peer, hashes := range request {
log.Trace("Fetching scheduled bodies", "peer", peer, "list", hashes)
↗ // Create a closure of the fetch and schedule in on a new thread
if f.completingHook != nil {
f.completingHook(hashes)
}
fetchBodies := f.completing[hashes[0]].fetchBodies
bodyFetchMeter.Mark(int64(len(hashes)))
↗ go func(peer string, hashes []common.Hash) {
resCh := make(chan *eth.Response)
↗ req, err := fetchBodies(hashes, resCh)
if err != nil {
return // Legacy code, yolo
}
defer req.Close()
↗ timeout := time.NewTimer(2 * fetchTimeout) // 2x leeway before dropping the peer
defer timeout.Stop()
↗ select {
case res := <-resCh:
res.Done <- nil
// Ignoring withdrawals here, since the block fetcher is not used post-merge.
≈ txs, uncles := res.Res.(*eth.BlockBodiesResponse).Unpack()
↗ f.FilterBodies(peer, txs, uncles, time.Now())
↗ case <-timeout.C:
// The peer didn't respond in time. The request
// was already rescheduled at this point, we were
// waiting for a catchup. With an unresponsive
// peer however, it's a protocol violation.
f.dropPeer(peer)
}
}(peer, hashes)
}
// Schedule the next fetch if blocks are still pending
f.rescheduleComplete(completeTimer)
↗ case filter := <-f.headerFilter:
// Headers arrived from a remote peer. Extract those that were explicitly
// requested by the fetcher, and return everything else so it's delivered
// to other parts of the system.
var task *headerFilterTask
select {
case task = <-filter:
case <-f.quit:
return
}
headerFilterInMeter.Mark(int64(len(task.headers)))
↗ // Split the batch of headers into unknown ones (to return to the caller),
// known incomplete ones (requiring body retrievals) and completed blocks.
unknown, incomplete, complete, lightHeaders := []*types.Header{}, []*blockAnnounce{}, []*types.Block{}, []*blockAnnounce{}
for _, header := range task.headers {
hash := header.Hash()
↗ // Filter fetcher-requested headers from other synchronisation algorithms
if announce := f.fetching[hash]; announce != nil && announce.origin == task.peer && f.fetched[hash] == nil && f.completing[hash] == nil && f.queued[hash] == nil {
// If the delivered header does not match the promised number, drop the announcer
if header.Number.Uint64() != announce.number {
log.Trace("Invalid block number fetched", "peer", announce.origin, "hash", header.Hash(), "announced", announce.number, "provided", header.Number)
f.dropPeer(announce.origin)
f.forgetHash(hash)
continue
}
// Collect all headers only if we are running in light
// mode and the headers are not imported by other means.
if f.light {
if f.getHeader(hash) == nil {
announce.header = header
lightHeaders = append(lightHeaders, announce)
}
f.forgetHash(hash)
continue
}
// Only keep if not imported by other means
if f.getBlock(hash) == nil {
announce.header = header
announce.time = task.time
↗ // If the block is empty (header only), short circuit into the final import queue
if header.TxHash == types.EmptyTxsHash && header.UncleHash == types.EmptyUncleHash {
log.Trace("Block empty, skipping body retrieval", "peer", announce.origin, "number", header.Number, "hash", header.Hash())
↗ block := types.NewBlockWithHeader(header)
block.ReceivedAt = task.time
↗ complete = append(complete, block)
f.completing[hash] = announce
continue
}
// Otherwise add to the list of blocks needing completion
incomplete = append(incomplete, announce)
} else {
log.Trace("Block already imported, discarding header", "peer", announce.origin, "number", header.Number, "hash", header.Hash())
f.forgetHash(hash)
}
} else {
// BlockFetcher doesn't know about it, add to the return list
unknown = append(unknown, header)
}
}
headerFilterOutMeter.Mark(int64(len(unknown)))
select {
case filter <- &headerFilterTask{headers: unknown, time: task.time}:
case <-f.quit:
return
}
// Schedule the retrieved headers for body completion
for _, announce := range incomplete {
hash := announce.header.Hash()
if _, ok := f.completing[hash]; ok {
continue
}
f.fetched[hash] = append(f.fetched[hash], announce)
if len(f.fetched) == 1 {
f.rescheduleComplete(completeTimer)
}
}
// Schedule the header for light fetcher import
for _, announce := range lightHeaders {
f.enqueue(announce.origin, announce.header, nil)
}
// Schedule the header-only blocks for import
for _, block := range complete {
if announce := f.completing[block.Hash()]; announce != nil {
f.enqueue(announce.origin, nil, block)
}
}
↗ case filter := <-f.bodyFilter:
// Block bodies arrived, extract any explicitly requested blocks, return the rest
var task *bodyFilterTask
select {
case task = <-filter:
case <-f.quit:
return
}
bodyFilterInMeter.Mark(int64(len(task.transactions)))
blocks := []*types.Block{}
// abort early if there's nothing explicitly requested
if len(f.completing) > 0 {
for i := 0; i < len(task.transactions) && i < len(task.uncles); i++ {
// Match up a body to any possible completion request
var (
matched = false
uncleHash common.Hash // calculated lazily and reused
txnHash common.Hash // calculated lazily and reused
)
for hash, announce := range f.completing {
if f.queued[hash] != nil || announce.origin != task.peer {
continue
}
if uncleHash == (common.Hash{}) {
uncleHash = types.CalcUncleHash(task.uncles[i])
}
if uncleHash != announce.header.UncleHash {
continue
}
if txnHash == (common.Hash{}) {
txnHash = types.DeriveSha(types.Transactions(task.transactions[i]), trie.NewStackTrie(nil))
}
if txnHash != announce.header.TxHash {
continue
}
// Mark the body matched, reassemble if still unknown
matched = true
if f.getBlock(hash) == nil {
≈ block := types.NewBlockWithHeader(announce.header).WithBody(types.Body{Transactions: task.transactions[i], Uncles: task.uncles[i]})
↗ block.ReceivedAt = task.time
blocks = append(blocks, block)
} else {
f.forgetHash(hash)
}
}
if matched {
task.transactions = append(task.transactions[:i], task.transactions[i+1:]...)
task.uncles = append(task.uncles[:i], task.uncles[i+1:]...)
i--
continue
}
}
}
bodyFilterOutMeter.Mark(int64(len(task.transactions)))
select {
case filter <- task:
case <-f.quit:
return
}
// Schedule the retrieved blocks for ordered import
for _, block := range blocks {
if announce := f.completing[block.Hash()]; announce != nil {
f.enqueue(announce.origin, nil, block)
}
}
}
}
}
core-geth validation — +3 -3
| | | |---|---| | File | [`block_fetcher.go`](https://github.com/etclabscore/core-geth/blob/v1.12.20/eth/fetcher/block_fetcher.go) | | Symbol | `BlockFetcher.loop` | | Ref | `v1.12.20` |--- a/core-geth/eth/fetcher/block_fetcher.go
+++ b/etc/eth/fetcher/block_fetcher_pow.go
@@ -131,7 +131,7 @@
log.Trace("Fetching scheduled headers", "peer", peer, "list", hashes)
// Create a closure of the fetch and schedule in on a new thread
- fetchHeader, hashes := f.fetching[hashes[0]].fetchHeader, hashes
+ fetchHeader := f.fetching[hashes[0]].fetchHeader
go func(peer string) {
if f.fetchingHook != nil {
f.fetchingHook(hashes)
@@ -211,7 +211,7 @@
case res := <-resCh:
res.Done <- nil
// Ignoring withdrawals here, since the block fetcher is not used post-merge.
- txs, uncles, _ := res.Res.(*eth.BlockBodiesResponse).Unpack()
+ txs, uncles := res.Res.(*eth.BlockBodiesResponse).Unpack()
f.FilterBodies(peer, txs, uncles, time.Now())
case <-timeout.C:
@@ -356,7 +356,7 @@
// Mark the body matched, reassemble if still unknown
matched = true
if f.getBlock(hash) == nil {
- block := types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i])
+ block := types.NewBlockWithHeader(announce.header).WithBody(types.Body{Transactions: task.transactions[i], Uncles: task.uncles[i]})
block.ReceivedAt = task.time
blocks = append(blocks, block)
} else {