mirror of
https://gitlab.com/pulsechaincom/erigon-pulse.git
synced 2024-12-25 04:57:17 +00:00
c4678ffd77
Properly ignore blocks coming from peers not in our peer list (blocked) and do never request anything from bad peers. Added some checks to account for blocks known when requesting hashes (missing parents).
430 lines
12 KiB
Go
430 lines
12 KiB
Go
package downloader
|
|
|
|
import (
|
|
"math"
|
|
"math/big"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/common"
|
|
"github.com/ethereum/go-ethereum/core"
|
|
"github.com/ethereum/go-ethereum/core/types"
|
|
"github.com/ethereum/go-ethereum/logger"
|
|
"github.com/ethereum/go-ethereum/logger/glog"
|
|
"gopkg.in/fatih/set.v0"
|
|
)
|
|
|
|
const (
|
|
maxBlockFetch = 256 // Amount of max blocks to be fetched per chunk
|
|
minDesiredPeerCount = 3 // Amount of peers desired to start syncing
|
|
)
|
|
|
|
type hashCheckFn func(common.Hash) bool
|
|
type chainInsertFn func(types.Blocks) error
|
|
type hashIterFn func() (common.Hash, error)
|
|
type currentTdFn func() *big.Int
|
|
|
|
type Downloader struct {
|
|
mu sync.RWMutex
|
|
queue *queue
|
|
peers peers
|
|
|
|
// Callbacks
|
|
hasBlock hashCheckFn
|
|
insertChain chainInsertFn
|
|
currentTd currentTdFn
|
|
|
|
// Status
|
|
fetchingHashes int32
|
|
downloadingBlocks int32
|
|
processingBlocks int32
|
|
|
|
// Channels
|
|
newPeerCh chan *peer
|
|
syncCh chan syncPack
|
|
HashCh chan []common.Hash
|
|
blockCh chan blockPack
|
|
quit chan struct{}
|
|
}
|
|
|
|
type blockPack struct {
|
|
peerId string
|
|
blocks []*types.Block
|
|
}
|
|
|
|
type syncPack struct {
|
|
peer *peer
|
|
hash common.Hash
|
|
ignoreInitial bool
|
|
}
|
|
|
|
func New(hasBlock hashCheckFn, insertChain chainInsertFn, currentTd currentTdFn) *Downloader {
|
|
downloader := &Downloader{
|
|
queue: newqueue(),
|
|
peers: make(peers),
|
|
hasBlock: hasBlock,
|
|
insertChain: insertChain,
|
|
currentTd: currentTd,
|
|
newPeerCh: make(chan *peer, 1),
|
|
syncCh: make(chan syncPack, 1),
|
|
HashCh: make(chan []common.Hash, 1),
|
|
blockCh: make(chan blockPack, 1),
|
|
quit: make(chan struct{}),
|
|
}
|
|
go downloader.peerHandler()
|
|
go downloader.update()
|
|
|
|
return downloader
|
|
}
|
|
|
|
func (d *Downloader) RegisterPeer(id string, td *big.Int, hash common.Hash, getHashes hashFetcherFn, getBlocks blockFetcherFn) error {
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
|
|
glog.V(logger.Detail).Infoln("Register peer", id)
|
|
|
|
// Create a new peer and add it to the list of known peers
|
|
peer := newPeer(id, td, hash, getHashes, getBlocks)
|
|
// add peer to our peer set
|
|
d.peers[id] = peer
|
|
// broadcast new peer
|
|
d.newPeerCh <- peer
|
|
|
|
return nil
|
|
}
|
|
|
|
func (d *Downloader) UnregisterPeer(id string) {
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
|
|
glog.V(logger.Detail).Infoln("Unregister peer", id)
|
|
|
|
delete(d.peers, id)
|
|
}
|
|
|
|
func (d *Downloader) peerHandler() {
|
|
// itimer is used to determine when to start ignoring `minDesiredPeerCount`
|
|
//itimer := time.NewTicker(5 * time.Second)
|
|
itimer := time.NewTimer(5 * time.Second)
|
|
out:
|
|
for {
|
|
select {
|
|
case <-d.newPeerCh:
|
|
itimer.Stop()
|
|
// Meet the `minDesiredPeerCount` before we select our best peer
|
|
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())
|
|
case <-d.quit:
|
|
break out
|
|
}
|
|
}
|
|
}
|
|
|
|
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()) {
|
|
// 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.
|
|
if p.td.Cmp(d.currentTd()) <= 0 || d.hasBlock(p.recentHash) {
|
|
return
|
|
}
|
|
|
|
glog.V(logger.Detail).Infoln("New peer with highest TD =", p.td)
|
|
d.syncCh <- syncPack{p, p.recentHash, false}
|
|
}
|
|
}
|
|
|
|
func (d *Downloader) update() {
|
|
out:
|
|
for {
|
|
select {
|
|
case sync := <-d.syncCh:
|
|
selectedPeer := sync.peer
|
|
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.
|
|
if err := d.startFetchingHashes(selectedPeer, sync.hash, sync.ignoreInitial); err != nil {
|
|
// handle error
|
|
glog.V(logger.Debug).Infoln("Error fetching hashes:", err)
|
|
// XXX Reset
|
|
break
|
|
}
|
|
|
|
// 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(selectedPeer); err != nil {
|
|
glog.V(logger.Debug).Infoln("Error downloading blocks:", err)
|
|
// XXX reset
|
|
break
|
|
}
|
|
|
|
glog.V(logger.Detail).Infoln("Sync completed")
|
|
|
|
d.process()
|
|
case <-d.quit:
|
|
break out
|
|
}
|
|
}
|
|
}
|
|
|
|
// XXX Make synchronous
|
|
func (d *Downloader) startFetchingHashes(p *peer, hash common.Hash, ignoreInitial bool) error {
|
|
glog.V(logger.Debug).Infof("Downloading hashes (%x) from %s", hash.Bytes()[:4], p.id)
|
|
|
|
start := time.Now()
|
|
|
|
// We ignore the initial hash in some cases (e.g. we received a block without it's parent)
|
|
// In such circumstances we don't need to download the block so don't add it to the queue.
|
|
if !ignoreInitial {
|
|
// 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 {
|
|
select {
|
|
case hashes := <-d.HashCh:
|
|
var done bool // determines whether we're done fetching hashes (i.e. common hash found)
|
|
hashSet := set.New()
|
|
for _, hash := range hashes {
|
|
if d.hasBlock(hash) {
|
|
glog.V(logger.Debug).Infof("Found common hash %x\n", hash[:4])
|
|
|
|
done = true
|
|
break
|
|
}
|
|
|
|
hashSet.Add(hash)
|
|
}
|
|
d.queue.put(hashSet)
|
|
|
|
// Add hashes to the chunk set
|
|
// Check if we're done fetching
|
|
if !done && len(hashes) > 0 {
|
|
//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)
|
|
break out
|
|
}
|
|
}
|
|
}
|
|
glog.V(logger.Detail).Infof("Downloaded hashes (%d). Took %v\n", d.queue.hashPool.Size(), time.Since(start))
|
|
|
|
return nil
|
|
}
|
|
|
|
func (d *Downloader) startFetchingBlocks(p *peer) error {
|
|
glog.V(logger.Detail).Infoln("Downloading", d.queue.hashPool.Size(), "blocks")
|
|
atomic.StoreInt32(&d.downloadingBlocks, 1)
|
|
|
|
start := time.Now()
|
|
|
|
// default ticker for re-fetching blocks everynow and then
|
|
ticker := time.NewTicker(20 * time.Millisecond)
|
|
out:
|
|
for {
|
|
select {
|
|
case blockPack := <-d.blockCh:
|
|
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.
|
|
chunk := d.queue.get(peer, maxBlockFetch)
|
|
if chunk == nil {
|
|
continue
|
|
}
|
|
|
|
//fmt.Println("fetching for", peer.id)
|
|
// XXX make fetch blocking.
|
|
// Fetch the chunk and check for error. If the peer was somehow
|
|
// already fetching a chunk due to a bug, it will be returned to
|
|
// the queue
|
|
if err := peer.fetch(chunk); err != nil {
|
|
// log for tracing
|
|
glog.V(logger.Debug).Infof("peer %s received double work (state = %v)\n", peer.id, peer.state)
|
|
d.queue.put(chunk.hashes)
|
|
}
|
|
}
|
|
atomic.StoreInt32(&d.downloadingBlocks, 1)
|
|
} else if len(d.queue.fetching) == 0 {
|
|
// 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
|
|
// to a `getBlocks` message. A timeout of 5 seconds is set. Peers
|
|
// that badly or poorly behave are removed from the peer set (not banned).
|
|
// Bad peers are excluded from the available peer set and therefor won't be
|
|
// reused. XXX We could re-introduce peers after X time.
|
|
d.queue.mu.Lock()
|
|
var badPeers []string
|
|
for pid, chunk := range d.queue.fetching {
|
|
if time.Since(chunk.itime) > 5*time.Second {
|
|
badPeers = append(badPeers, pid)
|
|
// remove peer as good peer from peer list
|
|
d.UnregisterPeer(pid)
|
|
}
|
|
}
|
|
d.queue.mu.Unlock()
|
|
|
|
for _, pid := range badPeers {
|
|
// A nil chunk is delivered so that the chunk's hashes are given
|
|
// back to the queue objects. When hashes are put back in the queue
|
|
// other (decent) peers can pick them up.
|
|
// XXX We could make use of a reputation system here ranking peers
|
|
// in their performance
|
|
// 1) Time for them to respond;
|
|
// 2) Measure their speed;
|
|
// 3) Amount and availability.
|
|
d.queue.deliver(pid, nil)
|
|
}
|
|
|
|
}
|
|
//fmt.Println(d.queue.hashPool.Size(), len(d.queue.fetching))
|
|
}
|
|
}
|
|
|
|
glog.V(logger.Detail).Infoln("Download blocks: done. Took", time.Since(start))
|
|
|
|
return nil
|
|
}
|
|
|
|
// Add an (unrequested) block to the downloader. This is usually done through the
|
|
// NewBlockMsg by the protocol handler.
|
|
func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) {
|
|
hash := block.Hash()
|
|
|
|
if d.hasBlock(hash) {
|
|
return
|
|
}
|
|
|
|
peer := d.peers.getPeer(id)
|
|
// if the peer is in our healthy list of peers; update the td
|
|
// and add the block. Otherwise just ignore it
|
|
if peer == nil {
|
|
glog.V(logger.Detail).Infof("Ignored block from bad peer %s\n", id)
|
|
return
|
|
}
|
|
|
|
peer.mu.Lock()
|
|
peer.td = td
|
|
peer.recentHash = block.Hash()
|
|
peer.mu.Unlock()
|
|
|
|
glog.V(logger.Detail).Infoln("Inserting new block from:", id)
|
|
d.queue.addBlock(id, block, td)
|
|
|
|
// if neither go ahead to process
|
|
if !(d.isFetchingHashes() || d.isDownloadingBlocks()) {
|
|
// Check if the parent of the received block is known.
|
|
// If the block is not know, request it otherwise, request.
|
|
phash := block.ParentHash()
|
|
if !d.hasBlock(phash) {
|
|
glog.V(logger.Detail).Infof("Missing parent %x, requires fetching\n", phash.Bytes()[:4])
|
|
d.syncCh <- syncPack{peer, peer.recentHash, true}
|
|
} else {
|
|
d.process()
|
|
}
|
|
}
|
|
}
|
|
|
|
// Deliver a chunk to the downloader. This is usually done through the BlocksMsg by
|
|
// the protocol handler.
|
|
func (d *Downloader) DeliverChunk(id string, blocks []*types.Block) {
|
|
d.blockCh <- blockPack{id, blocks}
|
|
}
|
|
|
|
func (d *Downloader) process() error {
|
|
atomic.StoreInt32(&d.processingBlocks, 1)
|
|
defer atomic.StoreInt32(&d.processingBlocks, 0)
|
|
|
|
// XXX this will move when optimised
|
|
// Sort the blocks by number. This bit needs much improvement. Right now
|
|
// it assumes full honesty form peers (i.e. it's not checked when the blocks
|
|
// link). We should at least check whihc queue match. This code could move
|
|
// to a seperate goroutine where it periodically checks for linked pieces.
|
|
types.BlockBy(types.Number).Sort(d.queue.blocks)
|
|
blocks := d.queue.blocks
|
|
|
|
glog.V(logger.Debug).Infoln("Inserting chain with", len(blocks), "blocks")
|
|
|
|
var err error
|
|
// Loop untill we're out of blocks
|
|
for len(blocks) != 0 {
|
|
max := int(math.Min(float64(len(blocks)), 256))
|
|
// TODO check for parent error. When there's a parent error we should stop
|
|
// processing and start requesting the `block.hash` so that it's parent and
|
|
// grandparents can be requested and queued.
|
|
err = d.insertChain(blocks[:max])
|
|
if err != nil && core.IsParentErr(err) {
|
|
glog.V(logger.Debug).Infoln("Aborting process due to missing parent. Fetching hashes")
|
|
|
|
// TODO change this. This shite
|
|
for i, block := range blocks[:max] {
|
|
if !d.hasBlock(block.ParentHash()) {
|
|
d.syncCh <- syncPack{d.peers.bestPeer(), block.Hash(), true}
|
|
// remove processed blocks
|
|
blocks = blocks[i:]
|
|
|
|
break
|
|
}
|
|
}
|
|
break
|
|
}
|
|
blocks = blocks[max:]
|
|
}
|
|
|
|
// This will allow the GC to remove the in memory blocks
|
|
if len(blocks) == 0 {
|
|
d.queue.blocks = nil
|
|
} else {
|
|
d.queue.blocks = blocks
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (d *Downloader) isFetchingHashes() bool {
|
|
return atomic.LoadInt32(&d.fetchingHashes) == 1
|
|
}
|
|
|
|
func (d *Downloader) isDownloadingBlocks() bool {
|
|
return atomic.LoadInt32(&d.downloadingBlocks) == 1
|
|
}
|
|
|
|
func (d *Downloader) isProcessing() bool {
|
|
return atomic.LoadInt32(&d.processingBlocks) == 1
|
|
}
|