aboutsummaryrefslogtreecommitdiffstats
path: root/eth/sync.go
diff options
context:
space:
mode:
Diffstat (limited to 'eth/sync.go')
-rw-r--r--eth/sync.go145
1 files changed, 8 insertions, 137 deletions
diff --git a/eth/sync.go b/eth/sync.go
index 751bc1a2a..82abb725f 100644
--- a/eth/sync.go
+++ b/eth/sync.go
@@ -12,11 +12,8 @@ import (
)
const (
- forceSyncCycle = 10 * time.Second // Time interval to force syncs, even if few peers are available
- 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
+ forceSyncCycle = 10 * time.Second // Time interval to force syncs, even if few peers are available
+ minDesiredPeerCount = 5 // Amount of peers desired to start syncing
// 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.
@@ -119,140 +116,15 @@ 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)
- 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 {
- select {
- case notifications := <-pm.newHashCh:
- // 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 {
- if time.Since(announce.time) > notifyFetchTimeout {
- delete(pending, hash)
- }
- }
- // Check if any notified blocks failed to arrive
- 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
- }
- delete(announces, hash)
- }
- }
- if len(request) == 0 {
- break
- }
- // 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)
- go peer.requestBlocks(hashes)
- }
- request = make(map[*peer][]common.Hash)
-
- case filter := <-pm.newBlockCh:
- // Blocks arrived, extract any explicit fetches, return all else
- var blocks types.Blocks
- select {
- case blocks = <-filter:
- case <-pm.quitSync:
- return
- }
-
- explicit, download := []*types.Block{}, []*types.Block{}
- for _, block := range blocks {
- hash := block.Hash()
-
- // Filter explicitly requested blocks from hash announcements
- if _, ok := pending[hash]; ok {
- // Discard if already imported by other means
- if !pm.chainman.HasBlock(hash) {
- explicit = append(explicit, block)
- } else {
- delete(pending, hash)
- }
- } else {
- download = append(download, block)
- }
- }
-
- select {
- case filter <- download:
- case <-pm.quitSync:
- return
- }
- // 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 {
- // Drop the block if it surely cannot fit
- if pm.chainman.HasBlock(hash) || !pm.chainman.HasBlock(block.ParentHash()) {
- // delete(pending, hash) // if we drop, it will re-fetch it, wait for timeout?
- continue
- }
- // 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)
- return
- }
- }
- }()
- }
-
- case <-pm.quitSync:
- return
- }
- }
-}
-
// syncer is responsible for periodically synchronising with the network, both
-// downloading hashes and blocks as well as retrieving cached ones.
+// downloading hashes and blocks as well as handling the announcement handler.
func (pm *ProtocolManager) syncer() {
- // Abort any pending syncs if we terminate
+ // Start and ensure cleanup of sync mechanisms
+ pm.fetcher.Start()
+ defer pm.fetcher.Stop()
defer pm.downloader.Terminate()
+ // Wait for different events to fire synchronisation operations
forceSync := time.Tick(forceSyncCycle)
for {
select {
@@ -273,8 +145,7 @@ func (pm *ProtocolManager) syncer() {
}
}
-// 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.
+// synchronise tries to sync up our local block chain with a remote peer.
func (pm *ProtocolManager) synchronise(peer *peer) {
// Short circuit if no peers are available
if peer == nil {