forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

eth/fetcher: schedule missing parent block synchronously

Browse source
This commit is contained in:
Zergity 2019-08-15 15:07:09 +07:00
parent 9a05d3e761
commit 0736bc9562
No known key found for this signature in database
GPG key ID: AE827ADEE29FDBB4
1 changed files with 30 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
eth/fetcher/fetcher.go
View file

@ -289,6 +289,24 @@ func (f *Fetcher) loop() {
fetchTimer := time.NewTimer(0) fetchTimer := time.NewTimer(0)
completeTimer := time.NewTimer(0) completeTimer := time.NewTimer(0)
scheduleBlockAnnouncement := func(notification *announce) bool {
if _, ok := f.fetching[notification.hash]; ok {
return false
}
if _, ok := f.completing[notification.hash]; ok {
return false
}
f.announces[notification.origin]++
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)
}
return true
}
for { for {
// Clean up any expired block fetches // Clean up any expired block fetches
for hash, announce := range f.fetching { for hash, announce := range f.fetching {
@ -327,13 +345,17 @@ func (f *Fetcher) loop() {
log.Info("Peer lost while fetching for parent", "peer", op.origin) log.Info("Peer lost while fetching for parent", "peer", op.origin)
break break
} }
go func() { scheduled := scheduleBlockAnnouncement(&announce{
log.Info("Notify of parent hash", "hash", parentHash, "number", op.block.NumberU64()-1, "peer", op.origin) hash: parentHash,
err := f.Notify(op.origin, parentHash, op.block.NumberU64()-1, time.Now(), headerRequestFn, bodyRequestFn) number: op.block.NumberU64() - 1,
if err != nil { time: time.Now(),
log.Error("Unable to notify of parent block", "error", err) origin: op.origin,
} fetchHeader: headerRequestFn,
}() fetchBodies: bodyRequestFn,
})
if scheduled {
log.Trace("Remote parent hash scheduled", "number", op.block.NumberU64()-1, "hash", parentHash, "peer", op.origin)
}
break break
} }
f.insert(op.origin, op.block) f.insert(op.origin, op.block)
@ -363,20 +385,7 @@ func (f *Fetcher) loop() {
} }
} }
// All is well, schedule the announce if block's not yet downloading // All is well, schedule the announce if block's not yet downloading
if _, ok := f.fetching[notification.hash]; ok { scheduleBlockAnnouncement(notification)
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: case op := <-f.inject:
// A direct block insertion was requested, try and fill any pending gaps // A direct block insertion was requested, try and fill any pending gaps