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: eth: discard fetched blocks that don't fit (no goroutine) --- eth/sync.go | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) (limited to 'eth/sync.go') diff --git a/eth/sync.go b/eth/sync.go index a25d4d4fd..7a3b5fcd4 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 { -- cgit v1.2.3 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: eth: randomly fetch announced block (don't hammer origin) --- eth/sync.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) (limited to 'eth/sync.go') diff --git a/eth/sync.go b/eth/sync.go index 7a3b5fcd4..7817266f8 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 -- cgit v1.2.3 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: eth: dedup fetches to ensure no blocks are pulled twice --- eth/sync.go | 28 +++++++++++++++++++++++----- 1 file changed, 23 insertions(+), 5 deletions(-) (limited to 'eth/sync.go') 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) -- cgit v1.2.3 From 66d3dc8690e0aa551e7b35a17006a2135b51c9bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 11 Jun 2015 15:56:08 +0300 Subject: eth, eth/downloader: move peer removal into downloader --- eth/sync.go | 32 ++------------------------------ 1 file changed, 2 insertions(+), 30 deletions(-) (limited to 'eth/sync.go') diff --git a/eth/sync.go b/eth/sync.go index 8fee21d7b..b127ca979 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -8,7 +8,6 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p/discover" @@ -332,33 +331,6 @@ func (pm *ProtocolManager) synchronise(peer *peer) { if peer.Td().Cmp(pm.chainman.Td()) <= 0 { return } - // FIXME if we have the hash in our chain and the TD of the peer is - // much higher than ours, something is wrong with us or the peer. - // Check if the hash is on our own chain - head := peer.Head() - if pm.chainman.HasBlock(head) { - glog.V(logger.Debug).Infoln("Synchronisation canceled: head already known") - return - } - // Get the hashes from the peer (synchronously) - glog.V(logger.Detail).Infof("Attempting synchronisation: %v, 0x%x", peer.id, head) - - err := pm.downloader.Synchronise(peer.id, head) - switch err { - case nil: - glog.V(logger.Detail).Infof("Synchronisation completed") - - case downloader.ErrBusy: - glog.V(logger.Detail).Infof("Synchronisation already in progress") - - case downloader.ErrTimeout, downloader.ErrBadPeer, downloader.ErrEmptyHashSet, downloader.ErrInvalidChain, downloader.ErrCrossCheckFailed: - glog.V(logger.Debug).Infof("Removing peer %v: %v", peer.id, err) - pm.removePeer(peer.id) - - case downloader.ErrPendingQueue: - glog.V(logger.Debug).Infoln("Synchronisation aborted:", err) - - default: - glog.V(logger.Warn).Infof("Synchronisation failed: %v", err) - } + // Otherwise try to sync with the downloader + pm.downloader.Synchronise(peer.id, peer.Head()) } -- cgit v1.2.3 From fc7abd98865f3bdc6cc36258026db98a649cd577 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 12 Jun 2015 13:35:29 +0300 Subject: eth, eth/downloader: move block processing into the downlaoder --- eth/sync.go | 53 +++-------------------------------------------------- 1 file changed, 3 insertions(+), 50 deletions(-) (limited to 'eth/sync.go') diff --git a/eth/sync.go b/eth/sync.go index b127ca979..88a76805c 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -1,9 +1,7 @@ package eth import ( - "math" "math/rand" - "sync/atomic" "time" "github.com/ethereum/go-ethereum/common" @@ -15,12 +13,10 @@ import ( const ( forceSyncCycle = 10 * time.Second // Time interval to force syncs, even if few peers are available - blockProcCycle = 500 * time.Millisecond // Time interval to check for new blocks to process notifyCheckCycle = 100 * time.Millisecond // Time interval to allow hash notifies to fulfill before hard fetching notifyArriveTimeout = 500 * time.Millisecond // Time allowance before an announced block is explicitly requested notifyFetchTimeout = 5 * time.Second // Maximum alloted time to return an explicitly requested block minDesiredPeerCount = 5 // Amount of peers desired to start syncing - blockProcAmount = 256 // This is the target size for the packs of transactions sent by txsyncLoop. // A pack can get larger than this if a single transactions exceeds this size. @@ -254,10 +250,10 @@ func (pm *ProtocolManager) fetcher() { // syncer is responsible for periodically synchronising with the network, both // downloading hashes and blocks as well as retrieving cached ones. func (pm *ProtocolManager) syncer() { - forceSync := time.Tick(forceSyncCycle) - blockProc := time.Tick(blockProcCycle) - blockProcPend := int32(0) + // Abort any pending syncs if we terminate + defer pm.downloader.Cancel() + forceSync := time.Tick(forceSyncCycle) for { select { case <-pm.newPeerCh: @@ -271,55 +267,12 @@ func (pm *ProtocolManager) syncer() { // Force a sync even if not enough peers are present go pm.synchronise(pm.peers.BestPeer()) - case <-blockProc: - // Try to pull some blocks from the downloaded - if atomic.CompareAndSwapInt32(&blockProcPend, 0, 1) { - go func() { - pm.processBlocks() - atomic.StoreInt32(&blockProcPend, 0) - }() - } - case <-pm.quitSync: return } } } -// processBlocks retrieves downloaded blocks from the download cache and tries -// to construct the local block chain with it. Note, since the block retrieval -// order matters, access to this function *must* be synchronized/serialized. -func (pm *ProtocolManager) processBlocks() error { - pm.wg.Add(1) - defer pm.wg.Done() - - // Short circuit if no blocks are available for insertion - blocks := pm.downloader.TakeBlocks() - if len(blocks) == 0 { - return nil - } - glog.V(logger.Debug).Infof("Inserting chain with %d blocks (#%v - #%v)\n", len(blocks), blocks[0].RawBlock.Number(), blocks[len(blocks)-1].RawBlock.Number()) - - for len(blocks) != 0 && !pm.quit { - // Retrieve the first batch of blocks to insert - max := int(math.Min(float64(len(blocks)), float64(blockProcAmount))) - raw := make(types.Blocks, 0, max) - for _, block := range blocks[:max] { - raw = append(raw, block.RawBlock) - } - // Try to inset the blocks, drop the originating peer if there's an error - index, err := pm.chainman.InsertChain(raw) - if err != nil { - glog.V(logger.Debug).Infoln("Downloaded block import failed:", err) - pm.removePeer(blocks[index].OriginPeer) - pm.downloader.Cancel() - return err - } - blocks = blocks[max:] - } - return nil -} - // synchronise tries to sync up our local block chain with a remote peer, both // adding various sanity checks as well as wrapping it with various log entries. func (pm *ProtocolManager) synchronise(peer *peer) { -- cgit v1.2.3 From b240983e2bafcde1c5902ce3a196b22475412f16 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 15 Jun 2015 12:26:05 +0300 Subject: eth, eth/downloader: do async block fetches, add dl tests --- eth/sync.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'eth/sync.go') diff --git a/eth/sync.go b/eth/sync.go index 88a76805c..917fc0fce 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -171,7 +171,7 @@ func (pm *ProtocolManager) fetcher() { // Send out all block requests for peer, hashes := range request { glog.V(logger.Debug).Infof("Explicitly fetching %d blocks from %s", len(hashes), peer.id) - peer.requestBlocks(hashes) + go peer.requestBlocks(hashes) } request = make(map[*peer][]common.Hash) -- cgit v1.2.3 From aa250e228a7f2eec5d512d05eb042b75e2755d30 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 15 Jun 2015 15:18:31 +0300 Subject: eth: don't refetch non fitting blocks to avoid duplicates --- eth/sync.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'eth/sync.go') diff --git a/eth/sync.go b/eth/sync.go index 917fc0fce..a3b177a4d 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -214,7 +214,7 @@ func (pm *ProtocolManager) fetcher() { if announce := pending[hash]; announce != nil { // Drop the block if it surely cannot fit if pm.chainman.HasBlock(hash) || !pm.chainman.HasBlock(block.ParentHash()) { - delete(pending, hash) + // delete(pending, hash) // if we drop, it will re-fetch it, wait for timeout? continue } // Otherwise accumulate for import -- cgit v1.2.3