mirror/erigon-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/erigon-pulse.git synced 2025-01-11 13:30:05 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #1227 from karalabe/block-fetcher-optimisations

Browse Source 
eth: optimize the notification/explicit fetch mechanism
This commit is contained in:
Jeffrey Wilcke 2015-06-11 03:31:41 -07:00
commit 13bd452faf
1 changed files with 37 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
eth/sync.go
View File

@ -127,10 +127,11 @@ func (pm *ProtocolManager) txsyncLoop() {
// fetcher is responsible for collecting hash notifications, and periodically // fetcher is responsible for collecting hash notifications, and periodically
// checking all unknown ones and individually fetching them. // checking all unknown ones and individually fetching them.
func (pm *ProtocolManager) fetcher() { func (pm *ProtocolManager) fetcher() {
announces := make(map[common.Hash]*blockAnnounce) announces := make(map[common.Hash][]*blockAnnounce)
request := make(map[*peer][]common.Hash) request := make(map[*peer][]common.Hash)
pending := make(map[common.Hash]*blockAnnounce) pending := make(map[common.Hash]*blockAnnounce)
cycle := time.Tick(notifyCheckCycle) cycle := time.Tick(notifyCheckCycle)
done := make(chan common.Hash)
// Iterate the block fetching until a quit is requested // Iterate the block fetching until a quit is requested
for { for {
@ -139,9 +140,18 @@ func (pm *ProtocolManager) fetcher() {
// A batch of hashes the notified, schedule them for retrieval // A batch of hashes the notified, schedule them for retrieval
glog.V(logger.Debug).Infof("Scheduling %d hash announcements from %s", len(notifications), notifications[0].peer.id) glog.V(logger.Debug).Infof("Scheduling %d hash announcements from %s", len(notifications), notifications[0].peer.id)
for _, announce := range notifications { for _, announce := range notifications {
announces[announce.hash] = announce // Skip if it's already pending fetch
if _, ok := pending[announce.hash]; ok {
continue
}
// Otherwise queue up the peer as a potential source
announces[announce.hash] = append(announces[announce.hash], announce)
} }
case hash := <-done:
// A pending import finished, remove all traces
delete(pending, hash)
case <-cycle: case <-cycle:
// Clean up any expired block fetches // Clean up any expired block fetches
for hash, announce := range pending { for hash, announce := range pending {
@ -150,8 +160,9 @@ func (pm *ProtocolManager) fetcher() {
} }
} }
// Check if any notified blocks failed to arrive // Check if any notified blocks failed to arrive
for hash, announce := range announces { for hash, all := range announces {
if time.Since(announce.time) > notifyArriveTimeout { if time.Since(all[0].time) > notifyArriveTimeout {
announce := all[rand.Intn(len(all))]
if !pm.chainman.HasBlock(hash) { if !pm.chainman.HasBlock(hash) {
request[announce.peer] = append(request[announce.peer], hash) request[announce.peer] = append(request[announce.peer], hash)
pending[hash] = announce pending[hash] = announce
@ -200,24 +211,32 @@ func (pm *ProtocolManager) fetcher() {
case <-pm.quitSync: case <-pm.quitSync:
return return
} }
// If any explicit fetches were replied to, import them // Create a closure with the retrieved blocks and origin peers
if count := len(explicit); count > 0 { peers := make([]*peer, 0, len(explicit))
glog.V(logger.Debug).Infof("Importing %d explicitly fetched blocks", count) blocks = make([]*types.Block, 0, len(explicit))
for _, block := range explicit {
// Create a closure with the retrieved blocks and origin peers hash := block.Hash()
peers := make([]*peer, 0, count) if announce := pending[hash]; announce != nil {
blocks := make([]*types.Block, 0, count) // Drop the block if it surely cannot fit
for _, block := range explicit { if pm.chainman.HasBlock(hash) || !pm.chainman.HasBlock(block.ParentHash()) {
hash := block.Hash()
if announce := pending[hash]; announce != nil {
peers = append(peers, announce.peer)
blocks = append(blocks, block)
delete(pending, hash) delete(pending, hash)
continue
} }
// Otherwise accumulate for import
peers = append(peers, announce.peer)
blocks = append(blocks, block)
} }
// Run the importer on a new thread }
// If any explicit fetches were replied to, import them
if count := len(blocks); count > 0 {
glog.V(logger.Debug).Infof("Importing %d explicitly fetched blocks", len(blocks))
go func() { go func() {
// Make sure all hashes are cleaned up
for _, block := range blocks {
hash := block.Hash()
defer func() { done <- hash }()
}
// Try and actually import the blocks
for i := 0; i < len(blocks); i++ { for i := 0; i < len(blocks); i++ {
if err := pm.importBlock(peers[i], blocks[i], nil); err != nil { if err := pm.importBlock(peers[i], blocks[i], nil); err != nil {
glog.V(logger.Detail).Infof("Failed to import explicitly fetched block: %v", err) glog.V(logger.Detail).Infof("Failed to import explicitly fetched block: %v", err)