aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorPéter Szilágyi <peterke@gmail.com>2015-06-10 18:08:00 +0800
committerPéter Szilágyi <peterke@gmail.com>2015-06-11 00:47:59 +0800
commite61db7145a07fedc16fadc5c438462ad42a7461c (patch)
treeefa91033a235708d97ae5f0ce92b5815b021558f
parent355b1e3bb1c749d8ecb19e967db988a34aa36788 (diff)
downloaddexon-e61db7145a07fedc16fadc5c438462ad42a7461c.tar
dexon-e61db7145a07fedc16fadc5c438462ad42a7461c.tar.gz
dexon-e61db7145a07fedc16fadc5c438462ad42a7461c.tar.bz2
dexon-e61db7145a07fedc16fadc5c438462ad42a7461c.tar.lz
dexon-e61db7145a07fedc16fadc5c438462ad42a7461c.tar.xz
dexon-e61db7145a07fedc16fadc5c438462ad42a7461c.tar.zst
dexon-e61db7145a07fedc16fadc5c438462ad42a7461c.zip
eth: dedup fetches to ensure no blocks are pulled twice
-rw-r--r--eth/sync.go28
1 files changed, 23 insertions, 5 deletions
diff --git a/eth/sync.go b/eth/sync.go
index 7817266f8..8fee21d7b 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)