aboutsummaryrefslogtreecommitdiffstats
path: root/eth/downloader/downloader.go
diff options
context:
space:
mode:
Diffstat (limited to 'eth/downloader/downloader.go')
-rw-r--r--eth/downloader/downloader.go148
1 files changed, 123 insertions, 25 deletions
diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 4e795af6d..c71cfa684 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -1,6 +1,7 @@
package downloader
import (
+ "errors"
"math"
"math/big"
"sync"
@@ -16,8 +17,15 @@ import (
)
const (
- maxBlockFetch = 256 // Amount of max blocks to be fetched per chunk
- minDesiredPeerCount = 3 // Amount of peers desired to start syncing
+ maxBlockFetch = 256 // Amount of max blocks to be fetched per chunk
+ minDesiredPeerCount = 3 // Amount of peers desired to start syncing
+ blockTtl = 15 * time.Second // The amount of time it takes for a request to time out
+)
+
+var (
+ errLowTd = errors.New("peer's TD is too low")
+ errBusy = errors.New("busy")
+ errUnknownPeer = errors.New("peer's unknown or unhealthy")
)
type hashCheckFn func(common.Hash) bool
@@ -82,7 +90,7 @@ func (d *Downloader) RegisterPeer(id string, td *big.Int, hash common.Hash, getH
d.mu.Lock()
defer d.mu.Unlock()
- glog.V(logger.Detail).Infoln("Register peer", id)
+ glog.V(logger.Detail).Infoln("Register peer", id, "TD =", td)
// Create a new peer and add it to the list of known peers
peer := newPeer(id, td, hash, getHashes, getBlocks)
@@ -94,6 +102,7 @@ func (d *Downloader) RegisterPeer(id string, td *big.Int, hash common.Hash, getH
return nil
}
+// UnregisterPeer unregister's a peer. This will prevent any action from the specified peer.
func (d *Downloader) UnregisterPeer(id string) {
d.mu.Lock()
defer d.mu.Unlock()
@@ -103,6 +112,73 @@ func (d *Downloader) UnregisterPeer(id string) {
delete(d.peers, id)
}
+// SynchroniseWithPeer will select the peer and use it for synchronising. If an empty string is given
+// it will use the best peer possible and synchronise if it's TD is higher than our own. If any of the
+// checks fail an error will be returned. This method is synchronous
+func (d *Downloader) SynchroniseWithPeer(id string) (types.Blocks, error) {
+ // Check if we're busy
+ if d.isBusy() {
+ return nil, errBusy
+ }
+
+ // Attempt to select a peer. This can either be nothing, which returns, best peer
+ // or selected peer. If no peer could be found an error will be returned
+ var p *peer
+ if len(id) == 0 {
+ p = d.peers[id]
+ if p == nil {
+ return nil, errUnknownPeer
+ }
+ } else {
+ p = d.peers.bestPeer()
+ }
+
+ // Make sure our td is lower than the peer's td
+ if p.td.Cmp(d.currentTd()) <= 0 || d.hasBlock(p.recentHash) {
+ return nil, errLowTd
+ }
+
+ // Get the hash from the peer and initiate the downloading progress.
+ err := d.getFromPeer(p, p.recentHash, false)
+ if err != nil {
+ return nil, err
+ }
+
+ return d.queue.blocks, nil
+}
+
+// Synchronise will synchronise using the best peer.
+func (d *Downloader) Synchronise() (types.Blocks, error) {
+ return d.SynchroniseWithPeer("")
+}
+
+func (d *Downloader) getFromPeer(p *peer, hash common.Hash, ignoreInitial bool) error {
+ glog.V(logger.Detail).Infoln("Synchronising with the network using:", p.id)
+ // Start the fetcher. This will block the update entirely
+ // interupts need to be send to the appropriate channels
+ // respectively.
+ if err := d.startFetchingHashes(p, hash, ignoreInitial); err != nil {
+ // handle error
+ glog.V(logger.Debug).Infoln("Error fetching hashes:", err)
+ // XXX Reset
+ return err
+ }
+
+ // Start fetching blocks in paralel. The strategy is simple
+ // take any available peers, seserve a chunk for each peer available,
+ // let the peer deliver the chunkn and periodically check if a peer
+ // has timedout. When done downloading, process blocks.
+ if err := d.startFetchingBlocks(p); err != nil {
+ glog.V(logger.Debug).Infoln("Error downloading blocks:", err)
+ // XXX reset
+ return err
+ }
+
+ glog.V(logger.Detail).Infoln("Sync completed")
+
+ return nil
+}
+
func (d *Downloader) peerHandler() {
// itimer is used to determine when to start ignoring `minDesiredPeerCount`
//itimer := time.NewTicker(5 * time.Second)
@@ -116,11 +192,18 @@ out:
if len(d.peers) < minDesiredPeerCount {
break
}
+
d.selectPeer(d.peers.bestPeer())
case <-itimer.C:
// The timer will make sure that the downloader keeps an active state
// in which it attempts to always check the network for highest td peers
- d.selectPeer(d.peers.bestPeer())
+ // Either select the peer or restart the timer if no peers could
+ // be selected.
+ if peer := d.peers.bestPeer(); peer != nil {
+ d.selectPeer(d.peers.bestPeer())
+ } else {
+ itimer.Reset(5 * time.Second)
+ }
case <-d.quit:
break out
}
@@ -131,7 +214,7 @@ func (d *Downloader) selectPeer(p *peer) {
// Make sure it's doing neither. Once done we can restart the
// downloading process if the TD is higher. For now just get on
// with whatever is going on. This prevents unecessary switching.
- if !(d.isFetchingHashes() || d.isDownloadingBlocks() || d.isProcessing()) {
+ if !d.isBusy() {
// selected peer must be better than our own
// XXX we also check the peer's recent hash to make sure we
// don't have it. Some peers report (i think) incorrect TD.
@@ -142,6 +225,7 @@ func (d *Downloader) selectPeer(p *peer) {
glog.V(logger.Detail).Infoln("New peer with highest TD =", p.td)
d.syncCh <- syncPack{p, p.recentHash, false}
}
+
}
func (d *Downloader) update() {
@@ -150,7 +234,7 @@ out:
select {
case sync := <-d.syncCh:
selectedPeer := sync.peer
- glog.V(logger.Detail).Infoln("Synchronising with network using:", selectedPeer.id)
+ glog.V(logger.Detail).Infoln("Synchronising with the network using:", selectedPeer.id)
// Start the fetcher. This will block the update entirely
// interupts need to be send to the appropriate channels
// respectively.
@@ -182,6 +266,9 @@ out:
// XXX Make synchronous
func (d *Downloader) startFetchingHashes(p *peer, hash common.Hash, ignoreInitial bool) error {
+ atomic.StoreInt32(&d.fetchingHashes, 1)
+ defer atomic.StoreInt32(&d.fetchingHashes, 0)
+
glog.V(logger.Debug).Infof("Downloading hashes (%x) from %s", hash.Bytes()[:4], p.id)
start := time.Now()
@@ -192,10 +279,8 @@ func (d *Downloader) startFetchingHashes(p *peer, hash common.Hash, ignoreInitia
// Add the hash to the queue first
d.queue.hashPool.Add(hash)
}
-
// Get the first batch of hashes
p.getHashes(hash)
- atomic.StoreInt32(&d.fetchingHashes, 1)
out:
for {
@@ -216,14 +301,16 @@ out:
d.queue.put(hashSet)
// Add hashes to the chunk set
- // Check if we're done fetching
- if !done && len(hashes) > 0 {
+ if len(hashes) == 0 { // Make sure the peer actually gave you something valid
+ glog.V(logger.Debug).Infof("Peer (%s) responded with empty hash set\n", p.id)
+ d.queue.reset()
+
+ break out
+ } else if !done { // Check if we're done fetching
//fmt.Println("re-fetch. current =", d.queue.hashPool.Size())
// Get the next set of hashes
p.getHashes(hashes[len(hashes)-1])
- atomic.StoreInt32(&d.fetchingHashes, 1)
- } else {
- atomic.StoreInt32(&d.fetchingHashes, 0)
+ } else { // we're done
break out
}
}
@@ -236,6 +323,7 @@ out:
func (d *Downloader) startFetchingBlocks(p *peer) error {
glog.V(logger.Detail).Infoln("Downloading", d.queue.hashPool.Size(), "blocks")
atomic.StoreInt32(&d.downloadingBlocks, 1)
+ defer atomic.StoreInt32(&d.downloadingBlocks, 0)
start := time.Now()
@@ -245,18 +333,18 @@ out:
for {
select {
case blockPack := <-d.blockCh:
- d.peers[blockPack.peerId].promote()
- d.queue.deliver(blockPack.peerId, blockPack.blocks)
- d.peers.setState(blockPack.peerId, idleState)
+ // If the peer was previously banned and failed to deliver it's pack
+ // in a reasonable time frame, ignore it's message.
+ if d.peers[blockPack.peerId] != nil {
+ d.peers[blockPack.peerId].promote()
+ d.queue.deliver(blockPack.peerId, blockPack.blocks)
+ d.peers.setState(blockPack.peerId, idleState)
+ }
case <-ticker.C:
// If there are unrequested hashes left start fetching
// from the available peers.
if d.queue.hashPool.Size() > 0 {
availablePeers := d.peers.get(idleState)
- if len(availablePeers) == 0 {
- glog.V(logger.Detail).Infoln("No peers available out of", len(d.peers))
- }
-
for _, peer := range availablePeers {
// Get a possible chunk. If nil is returned no chunk
// could be returned due to no hashes available.
@@ -281,8 +369,6 @@ out:
// When there are no more queue and no more `fetching`. We can
// safely assume we're done. Another part of the process will check
// for parent errors and will re-request anything that's missing
- atomic.StoreInt32(&d.downloadingBlocks, 0)
- // Break out so that we can process with processing blocks
break out
} else {
// Check for bad peers. Bad peers may indicate a peer not responding
@@ -293,7 +379,7 @@ out:
d.queue.mu.Lock()
var badPeers []string
for pid, chunk := range d.queue.fetching {
- if time.Since(chunk.itime) > 5*time.Second {
+ if time.Since(chunk.itime) > blockTtl {
badPeers = append(badPeers, pid)
// remove peer as good peer from peer list
d.UnregisterPeer(pid)
@@ -353,7 +439,7 @@ func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) {
d.queue.addBlock(id, block, td)
// if neither go ahead to process
- if !(d.isFetchingHashes() || d.isDownloadingBlocks()) {
+ if !d.isBusy() {
// Check if the parent of the received block is known.
// If the block is not know, request it otherwise, request.
phash := block.ParentHash()
@@ -383,8 +469,11 @@ func (d *Downloader) process() error {
// to a seperate goroutine where it periodically checks for linked pieces.
types.BlockBy(types.Number).Sort(d.queue.blocks)
blocks := d.queue.blocks
+ if len(blocks) == 0 {
+ return nil
+ }
- glog.V(logger.Debug).Infoln("Inserting chain with", len(blocks), "blocks")
+ glog.V(logger.Debug).Infof("Inserting chain with %d blocks (#%v - #%v)\n", len(blocks), blocks[0].Number(), blocks[len(blocks)-1].Number())
var err error
// Loop untill we're out of blocks
@@ -408,6 +497,11 @@ func (d *Downloader) process() error {
}
}
break
+ } else if err != nil {
+ // Reset chain completely. This needs much, much improvement.
+ // instead: check all blocks leading down to this block false block and remove it
+ blocks = nil
+ break
}
blocks = blocks[max:]
}
@@ -432,3 +526,7 @@ func (d *Downloader) isDownloadingBlocks() bool {
func (d *Downloader) isProcessing() bool {
return atomic.LoadInt32(&d.processingBlocks) == 1
}
+
+func (d *Downloader) isBusy() bool {
+ return d.isFetchingHashes() || d.isDownloadingBlocks() || d.isProcessing()
+}
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-26 01:29:53 +0800