From b9affbf9fe1ca8e76aef5efdc456a43dff38dc8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 10 Jun 2015 12:17:06 +0300 Subject: [PATCH 1/3] eth: discard fetched blocks that don't fit (no goroutine) --- eth/sync.go | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/eth/sync.go b/eth/sync.go index a25d4d4fdf..7a3b5fcd4b 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -200,23 +200,23 @@ func (pm *ProtocolManager) fetcher() { case <-pm.quitSync: return } - // If any explicit fetches were replied to, import them - if count := len(explicit); count > 0 { - glog.V(logger.Debug).Infof("Importing %d explicitly fetched blocks", count) - - // Create a closure with the retrieved blocks and origin peers - peers := make([]*peer, 0, count) - blocks := make([]*types.Block, 0, count) - for _, block := range explicit { - hash := block.Hash() - if announce := pending[hash]; announce != nil { + // Create a closure with the retrieved blocks and origin peers + peers := make([]*peer, 0, len(explicit)) + blocks = make([]*types.Block, 0, len(explicit)) + for _, block := range explicit { + hash := block.Hash() + if announce := pending[hash]; announce != nil { + // Filter out blocks too new to import anyway + if !pm.chainman.HasBlock(hash) && pm.chainman.HasBlock(block.ParentHash()) { peers = append(peers, announce.peer) blocks = append(blocks, block) - - delete(pending, hash) } + delete(pending, hash) } - // 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() { for i := 0; i < len(blocks); i++ { if err := pm.importBlock(peers[i], blocks[i], nil); err != nil { From 355b1e3bb1c749d8ecb19e967db988a34aa36788 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 10 Jun 2015 12:30:35 +0300 Subject: [PATCH 2/3] eth: randomly fetch announced block (don't hammer origin) --- eth/sync.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/eth/sync.go b/eth/sync.go index 7a3b5fcd4b..7817266f8d 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -127,7 +127,7 @@ func (pm *ProtocolManager) txsyncLoop() { // fetcher is responsible for collecting hash notifications, and periodically // checking all unknown ones and individually fetching them. func (pm *ProtocolManager) fetcher() { - announces := make(map[common.Hash]*blockAnnounce) + announces := make(map[common.Hash][]*blockAnnounce) request := make(map[*peer][]common.Hash) pending := make(map[common.Hash]*blockAnnounce) cycle := time.Tick(notifyCheckCycle) @@ -139,7 +139,7 @@ func (pm *ProtocolManager) fetcher() { // 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) for _, announce := range notifications { - announces[announce.hash] = announce + announces[announce.hash] = append(announces[announce.hash], announce) } case <-cycle: @@ -150,8 +150,9 @@ func (pm *ProtocolManager) fetcher() { } } // Check if any notified blocks failed to arrive - for hash, announce := range announces { - if time.Since(announce.time) > notifyArriveTimeout { + for hash, all := range announces { + if time.Since(all[0].time) > notifyArriveTimeout { + announce := all[rand.Intn(len(all))] if !pm.chainman.HasBlock(hash) { request[announce.peer] = append(request[announce.peer], hash) pending[hash] = announce From e61db7145a07fedc16fadc5c438462ad42a7461c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 10 Jun 2015 13:08:00 +0300 Subject: [PATCH 3/3] eth: dedup fetches to ensure no blocks are pulled twice --- eth/sync.go | 28 +++++++++++++++++++++++----- 1 file changed, 23 insertions(+), 5 deletions(-) diff --git a/eth/sync.go b/eth/sync.go index 7817266f8d..8fee21d7b4 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -131,6 +131,7 @@ func (pm *ProtocolManager) fetcher() { request := make(map[*peer][]common.Hash) pending := make(map[common.Hash]*blockAnnounce) cycle := time.Tick(notifyCheckCycle) + done := make(chan common.Hash) // Iterate the block fetching until a quit is requested for { @@ -139,9 +140,18 @@ func (pm *ProtocolManager) fetcher() { // 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) for _, announce := range notifications { + // 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: // Clean up any expired block fetches for hash, announce := range pending { @@ -207,18 +217,26 @@ func (pm *ProtocolManager) fetcher() { for _, block := range explicit { hash := block.Hash() if announce := pending[hash]; announce != nil { - // Filter out blocks too new to import anyway - if !pm.chainman.HasBlock(hash) && pm.chainman.HasBlock(block.ParentHash()) { - peers = append(peers, announce.peer) - blocks = append(blocks, block) + // Drop the block if it surely cannot fit + if pm.chainman.HasBlock(hash) || !pm.chainman.HasBlock(block.ParentHash()) { + delete(pending, hash) + continue } - delete(pending, hash) + // Otherwise accumulate for import + peers = append(peers, announce.peer) + blocks = append(blocks, block) } } // 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() { + // 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++ { if err := pm.importBlock(peers[i], blocks[i], nil); err != nil { glog.V(logger.Detail).Infof("Failed to import explicitly fetched block: %v", err)