eth/fetcher: use iter.seq in blobfetcher

This commit is contained in:
healthykim 2026-06-18 18:29:13 +02:00
parent 999e6013a3
commit 294ac215af

View file

@ -17,6 +17,7 @@
package fetcher package fetcher
import ( import (
"iter"
"math/rand" "math/rand"
"slices" "slices"
"sort" "sort"
@ -750,15 +751,15 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
wasIdle := len(f.requests) == 0 wasIdle := len(f.requests) == 0
// For each active peer, try to schedule some payload fetches. // For each active peer, try to schedule some payload fetches.
f.forEachPeer(actives, func(peer string) { for peer := range f.peers(actives) {
if len(f.announces[peer]) == 0 || len(f.requests[peer]) != 0 { if len(f.announces[peer]) == 0 || len(f.requests[peer]) != 0 {
return // continue continue
} }
var ( var (
hashes []common.Hash hashes []common.Hash
custodies []types.CustodyBitmap custodies []types.CustodyBitmap
) )
f.forEachAnnounce(f.announces[peer], func(hash common.Hash, cells types.CustodyBitmap) bool { for hash, cells := range f.announcesByArrival(f.announces[peer]) {
var unfetched types.CustodyBitmap var unfetched types.CustodyBitmap
if f.fetches[hash] == nil { if f.fetches[hash] == nil {
// tx is not being fetched // tx is not being fetched
@ -793,8 +794,11 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
f.alternates[hash][peer] = cells f.alternates[hash][peer] = cells
} }
return len(hashes) < maxPayloadRetrievals // Stop once we've accumulated enough hashes for this peer
}) if len(hashes) >= maxPayloadRetrievals {
break
}
}
// If any hashes were allocated, request them from the peer // If any hashes were allocated, request them from the peer
if len(hashes) > 0 { if len(hashes) > 0 {
@ -829,7 +833,7 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
} }
}() }()
} }
}) }
// If a new request was fired, schedule a timeout timer // If a new request was fired, schedule a timeout timer
if wasIdle && len(f.requests) > 0 { if wasIdle && len(f.requests) > 0 {
@ -837,48 +841,57 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
} }
} }
// forEachAnnounce loops over the given announcements in arrival order, invoking // announcesByArrival returns an iterator over the given announcements
// the do function for each until it returns false. We enforce an arrival // in arrival order. We enforce an arrival ordering to minimize
// ordering to minimize the chances of transaction nonce-gaps, which result in // the chances of transaction nonce-gaps, which result in
// transactions being rejected by the txpool. // transactions being rejected by the txpool.
func (f *BlobFetcher) forEachAnnounce(announces map[common.Hash]*cellWithSeq, do func(hash common.Hash, cells types.CustodyBitmap) bool) {
type announcement struct { func (f *BlobFetcher) announcesByArrival(announces map[common.Hash]*cellWithSeq) iter.Seq2[common.Hash, types.CustodyBitmap] {
hash common.Hash return func(yield func(hash common.Hash, cells types.CustodyBitmap) bool) {
cells types.CustodyBitmap type announcement struct {
seq uint64 hash common.Hash
} cells types.CustodyBitmap
// Process announcements by their arrival order seq uint64
list := make([]announcement, 0, len(announces)) }
for hash, entry := range announces { // Process announcements by their arrival order
list = append(list, announcement{hash: hash, cells: entry.cells, seq: entry.seq}) list := make([]announcement, 0, len(announces))
} for hash, entry := range announces {
sort.Slice(list, func(i, j int) bool { list = append(list, announcement{hash: hash, cells: entry.cells, seq: entry.seq})
return list[i].seq < list[j].seq }
}) sort.Slice(list, func(i, j int) bool {
for i := range list { return list[i].seq < list[j].seq
if !do(list[i].hash, list[i].cells) { })
return for i := range list {
if !yield(list[i].hash, list[i].cells) {
return
}
} }
} }
} }
// forEachPeer does a range loop over a map of peers in production, but during // peers returns an iterator over a map of peers in production, but during
// testing it does a deterministic sorted random to allow reproducing issues. // testing it does a deterministic sorted random to allow reproducing issues.
func (f *BlobFetcher) forEachPeer(peers map[string]struct{}, do func(peer string)) { func (f *BlobFetcher) peers(peers map[string]struct{}) iter.Seq[string] {
// If we're running production(step == nil), use whatever Go's map gives us return func(yield func(peer string) bool) {
if f.step == nil { // If we're running production(step == nil), use whatever Go's map gives us
for peer := range peers { if f.step == nil {
do(peer) for peer := range peers {
if !yield(peer) {
return
}
}
return
}
// We're running the test suite, make iteration deterministic (sorted by peer id)
list := make([]string, 0, len(peers))
for peer := range peers {
list = append(list, peer)
}
sort.Strings(list)
for _, peer := range list {
if !yield(peer) {
return
}
} }
return
}
// We're running the test suite, make iteration deterministic (sorted by peer id)
list := make([]string, 0, len(peers))
for peer := range peers {
list = append(list, peer)
}
sort.Strings(list)
for _, peer := range list {
do(peer)
} }
} }