eth/fetcher, trie: unit test reliability fixes #23020 #23415 (#1070)

Some tests take quite some time during exit, which I think causes
some appveyor fails like this:

    https://ci.appveyor.com/project/ethereum/go-ethereum/builds/39511210/job/xhom84eg2e4uulq3

One of the things that seem to take time during exit is waiting
(up to 100ms) for the syncbloom to close. This PR changes it to use
a channel, instead of looping with a 100ms wait.

This also includes some unrelated changes improving the reliability of
eth/fetcher tests, which fail a lot because they are time-dependent.

Co-authored-by: Martin Holst Swende <martin@swende.se>
This commit is contained in:
Daniel Liu 2025-06-17 13:21:18 +08:00 committed by GitHub
parent a41a92978d
commit 5e8a096844
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 46 additions and 24 deletions

View file

@ -753,15 +753,17 @@ func (f *Fetcher) insert(peer string, block *types.Block) {
// internal state. // internal state.
func (f *Fetcher) forgetHash(hash common.Hash) { func (f *Fetcher) forgetHash(hash common.Hash) {
// Remove all pending announces and decrement DOS counters // Remove all pending announces and decrement DOS counters
for _, announce := range f.announced[hash] { if announceMap, ok := f.announced[hash]; ok {
f.announces[announce.origin]-- for _, announce := range announceMap {
if f.announces[announce.origin] == 0 { f.announces[announce.origin]--
delete(f.announces, announce.origin) if f.announces[announce.origin] <= 0 {
delete(f.announces, announce.origin)
}
}
delete(f.announced, hash)
if f.announceChangeHook != nil {
f.announceChangeHook(hash, false)
} }
}
delete(f.announced, hash)
if f.announceChangeHook != nil {
f.announceChangeHook(hash, false)
} }
// Remove any pending fetches and decrement the DOS counters // Remove any pending fetches and decrement the DOS counters
if announce := f.fetching[hash]; announce != nil { if announce := f.fetching[hash]; announce != nil {

View file

@ -677,15 +677,30 @@ func testInvalidNumberAnnouncement(t *testing.T, protocol int) {
badBodyFetcher := tester.makeBodyFetcher("bad", blocks, 0) badBodyFetcher := tester.makeBodyFetcher("bad", blocks, 0)
imported := make(chan *types.Block) imported := make(chan *types.Block)
announced := make(chan interface{})
tester.fetcher.signHook = func(block *types.Block) error { tester.fetcher.signHook = func(block *types.Block) error {
imported <- block imported <- block
return nil return nil
} }
// Announce a block with a bad number, check for immediate drop // Announce a block with a bad number, check for immediate drop
tester.fetcher.announceChangeHook = func(hash common.Hash, b bool) {
announced <- nil
}
tester.fetcher.Notify("bad", hashes[0], 2, time.Now().Add(-arriveTimeout), badHeaderFetcher, badBodyFetcher) tester.fetcher.Notify("bad", hashes[0], 2, time.Now().Add(-arriveTimeout), badHeaderFetcher, badBodyFetcher)
verifyAnnounce := func() {
for i := 0; i < 2; i++ {
select {
case <-announced:
continue
case <-time.After(1 * time.Second):
t.Fatal("announce timeout")
return
}
}
}
verifyAnnounce()
verifyImportEvent(t, imported, false) verifyImportEvent(t, imported, false)
tester.lock.RLock() tester.lock.RLock()
dropped := tester.drops["bad"] dropped := tester.drops["bad"]
tester.lock.RUnlock() tester.lock.RUnlock()
@ -698,6 +713,7 @@ func testInvalidNumberAnnouncement(t *testing.T, protocol int) {
goodBodyFetcher := tester.makeBodyFetcher("good", blocks, 0) goodBodyFetcher := tester.makeBodyFetcher("good", blocks, 0)
// Make sure a good announcement passes without a drop // Make sure a good announcement passes without a drop
tester.fetcher.Notify("good", hashes[0], 1, time.Now().Add(-arriveTimeout), goodHeaderFetcher, goodBodyFetcher) tester.fetcher.Notify("good", hashes[0], 1, time.Now().Add(-arriveTimeout), goodHeaderFetcher, goodBodyFetcher)
verifyAnnounce()
verifyImportEvent(t, imported, true) verifyImportEvent(t, imported, true)
tester.lock.RLock() tester.lock.RLock()

View file

@ -45,11 +45,12 @@ var (
// provided disk database on creation in a background thread and will only start // provided disk database on creation in a background thread and will only start
// returning live results once that's finished. // returning live results once that's finished.
type SyncBloom struct { type SyncBloom struct {
bloom *bloomfilter.Filter bloom *bloomfilter.Filter
inited uint32 inited uint32
closer sync.Once closer sync.Once
closed uint32 closed uint32
pend sync.WaitGroup pend sync.WaitGroup
closeCh chan struct{}
} }
// NewSyncBloom creates a new bloom filter of the given size (in megabytes) and // NewSyncBloom creates a new bloom filter of the given size (in megabytes) and
@ -64,7 +65,8 @@ func NewSyncBloom(memory uint64, database ethdb.Iteratee) *SyncBloom {
// Assemble the fast sync bloom and init it from previous sessions // Assemble the fast sync bloom and init it from previous sessions
b := &SyncBloom{ b := &SyncBloom{
bloom: bloom, bloom: bloom,
closeCh: make(chan struct{}),
} }
b.pend.Add(2) b.pend.Add(2)
go func() { go func() {
@ -125,16 +127,17 @@ func (b *SyncBloom) init(database ethdb.Iteratee) {
// meter periodically recalculates the false positive error rate of the bloom // meter periodically recalculates the false positive error rate of the bloom
// filter and reports it in a metric. // filter and reports it in a metric.
func (b *SyncBloom) meter() { func (b *SyncBloom) meter() {
for { // check every second
// Report the current error ration. No floats, lame, scale it up. tick := time.NewTicker(1 * time.Second)
bloomErrorGauge.Update(int64(b.bloom.FalsePosititveProbability() * 100000)) defer tick.Stop()
// Wait one second, but check termination more frequently for {
for i := 0; i < 10; i++ { select {
if atomic.LoadUint32(&b.closed) == 1 { case <-tick.C:
return // Report the current error ration. No floats, lame, scale it up.
} bloomErrorGauge.Update(int64(b.bloom.FalsePosititveProbability() * 100000))
time.Sleep(100 * time.Millisecond) case <-b.closeCh:
return
} }
} }
} }
@ -145,6 +148,7 @@ func (b *SyncBloom) Close() error {
b.closer.Do(func() { b.closer.Do(func() {
// Ensure the initializer is stopped // Ensure the initializer is stopped
atomic.StoreUint32(&b.closed, 1) atomic.StoreUint32(&b.closed, 1)
close(b.closeCh)
b.pend.Wait() b.pend.Wait()
// Wipe the bloom, but mark it "uninited" just in case someone attempts an access // Wipe the bloom, but mark it "uninited" just in case someone attempts an access