f8d0cd9906
Not sure if this is the right approach. Why? BlockChain shouldn't need the "Ethereum" object. BlockChain shouldn't need to worry about notifying listeners or message propagation.
348 lines
8.1 KiB
Go
348 lines
8.1 KiB
Go
package eth
|
|
|
|
import (
|
|
"bytes"
|
|
"container/list"
|
|
"fmt"
|
|
"math"
|
|
"math/big"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/chain"
|
|
"github.com/ethereum/go-ethereum/chain/types"
|
|
"github.com/ethereum/go-ethereum/ethutil"
|
|
"github.com/ethereum/go-ethereum/logger"
|
|
"github.com/ethereum/go-ethereum/state"
|
|
"github.com/ethereum/go-ethereum/wire"
|
|
)
|
|
|
|
var poollogger = logger.NewLogger("BPOOL")
|
|
|
|
type block struct {
|
|
from *Peer
|
|
peer *Peer
|
|
block *types.Block
|
|
reqAt time.Time
|
|
requested int
|
|
}
|
|
|
|
type BlockPool struct {
|
|
mut sync.Mutex
|
|
|
|
eth *Ethereum
|
|
|
|
hashes [][]byte
|
|
pool map[string]*block
|
|
|
|
td *big.Int
|
|
quit chan bool
|
|
|
|
fetchingHashes bool
|
|
downloadStartedAt time.Time
|
|
|
|
ChainLength, BlocksProcessed int
|
|
|
|
peer *Peer
|
|
}
|
|
|
|
func NewBlockPool(eth *Ethereum) *BlockPool {
|
|
return &BlockPool{
|
|
eth: eth,
|
|
pool: make(map[string]*block),
|
|
td: ethutil.Big0,
|
|
quit: make(chan bool),
|
|
}
|
|
}
|
|
|
|
func (self *BlockPool) Len() int {
|
|
return len(self.hashes)
|
|
}
|
|
|
|
func (self *BlockPool) Reset() {
|
|
self.pool = make(map[string]*block)
|
|
self.hashes = nil
|
|
}
|
|
|
|
func (self *BlockPool) HasLatestHash() bool {
|
|
self.mut.Lock()
|
|
defer self.mut.Unlock()
|
|
|
|
return self.pool[string(self.eth.ChainManager().CurrentBlock.Hash())] != nil
|
|
}
|
|
|
|
func (self *BlockPool) HasCommonHash(hash []byte) bool {
|
|
return self.eth.ChainManager().GetBlock(hash) != nil
|
|
}
|
|
|
|
func (self *BlockPool) Blocks() (blocks types.Blocks) {
|
|
for _, item := range self.pool {
|
|
if item.block != nil {
|
|
blocks = append(blocks, item.block)
|
|
}
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
func (self *BlockPool) FetchHashes(peer *Peer) bool {
|
|
highestTd := self.eth.HighestTDPeer()
|
|
|
|
if (self.peer == nil && peer.td.Cmp(highestTd) >= 0) || (self.peer != nil && peer.td.Cmp(self.peer.td) > 0) || self.peer == peer {
|
|
if self.peer != peer {
|
|
poollogger.Debugf("Found better suitable peer (%v vs %v)\n", self.td, peer.td)
|
|
|
|
if self.peer != nil {
|
|
self.peer.doneFetchingHashes = true
|
|
}
|
|
}
|
|
|
|
self.peer = peer
|
|
self.td = peer.td
|
|
|
|
if !self.HasLatestHash() {
|
|
peer.doneFetchingHashes = false
|
|
|
|
const amount = 256
|
|
peerlogger.Debugf("Fetching hashes (%d) %x...\n", amount, peer.lastReceivedHash[0:4])
|
|
peer.QueueMessage(wire.NewMessage(wire.MsgGetBlockHashesTy, []interface{}{peer.lastReceivedHash, uint32(amount)}))
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func (self *BlockPool) AddHash(hash []byte, peer *Peer) {
|
|
self.mut.Lock()
|
|
defer self.mut.Unlock()
|
|
|
|
if self.pool[string(hash)] == nil {
|
|
self.pool[string(hash)] = &block{peer, nil, nil, time.Now(), 0}
|
|
|
|
self.hashes = append([][]byte{hash}, self.hashes...)
|
|
}
|
|
}
|
|
|
|
func (self *BlockPool) Add(b *types.Block, peer *Peer) {
|
|
self.addBlock(b, peer, false)
|
|
}
|
|
|
|
func (self *BlockPool) AddNew(b *types.Block, peer *Peer) {
|
|
self.addBlock(b, peer, true)
|
|
}
|
|
|
|
func (self *BlockPool) addBlock(b *types.Block, peer *Peer, newBlock bool) {
|
|
self.mut.Lock()
|
|
defer self.mut.Unlock()
|
|
|
|
hash := string(b.Hash())
|
|
|
|
if self.pool[hash] == nil && !self.eth.ChainManager().HasBlock(b.Hash()) {
|
|
poollogger.Infof("Got unrequested block (%x...)\n", hash[0:4])
|
|
|
|
self.hashes = append(self.hashes, b.Hash())
|
|
self.pool[hash] = &block{peer, peer, b, time.Now(), 0}
|
|
|
|
// The following is only performed on an unrequested new block
|
|
if newBlock {
|
|
fmt.Println("1.", !self.eth.ChainManager().HasBlock(b.PrevHash), ethutil.Bytes2Hex(b.Hash()[0:4]), ethutil.Bytes2Hex(b.PrevHash[0:4]))
|
|
fmt.Println("2.", self.pool[string(b.PrevHash)] == nil)
|
|
fmt.Println("3.", !self.fetchingHashes)
|
|
if !self.eth.ChainManager().HasBlock(b.PrevHash) && self.pool[string(b.PrevHash)] == nil && !self.fetchingHashes {
|
|
poollogger.Infof("Unknown chain, requesting (%x...)\n", b.PrevHash[0:4])
|
|
peer.QueueMessage(wire.NewMessage(wire.MsgGetBlockHashesTy, []interface{}{b.Hash(), uint32(256)}))
|
|
}
|
|
}
|
|
} else if self.pool[hash] != nil {
|
|
self.pool[hash].block = b
|
|
}
|
|
|
|
self.BlocksProcessed++
|
|
}
|
|
|
|
func (self *BlockPool) Remove(hash []byte) {
|
|
self.mut.Lock()
|
|
defer self.mut.Unlock()
|
|
|
|
self.hashes = ethutil.DeleteFromByteSlice(self.hashes, hash)
|
|
delete(self.pool, string(hash))
|
|
}
|
|
|
|
func (self *BlockPool) DistributeHashes() {
|
|
self.mut.Lock()
|
|
defer self.mut.Unlock()
|
|
|
|
var (
|
|
peerLen = self.eth.peers.Len()
|
|
amount = 256 * peerLen
|
|
dist = make(map[*Peer][][]byte)
|
|
)
|
|
|
|
num := int(math.Min(float64(amount), float64(len(self.pool))))
|
|
for i, j := 0, 0; i < len(self.hashes) && j < num; i++ {
|
|
hash := self.hashes[i]
|
|
item := self.pool[string(hash)]
|
|
|
|
if item != nil && item.block == nil {
|
|
var peer *Peer
|
|
lastFetchFailed := time.Since(item.reqAt) > 5*time.Second
|
|
|
|
// Handle failed requests
|
|
if lastFetchFailed && item.requested > 5 && item.peer != nil {
|
|
if item.requested < 100 {
|
|
// Select peer the hash was retrieved off
|
|
peer = item.from
|
|
} else {
|
|
// Remove it
|
|
self.hashes = ethutil.DeleteFromByteSlice(self.hashes, hash)
|
|
delete(self.pool, string(hash))
|
|
}
|
|
} else if lastFetchFailed || item.peer == nil {
|
|
// Find a suitable, available peer
|
|
eachPeer(self.eth.peers, func(p *Peer, v *list.Element) {
|
|
if peer == nil && len(dist[p]) < amount/peerLen && p.statusKnown {
|
|
peer = p
|
|
}
|
|
})
|
|
}
|
|
|
|
if peer != nil {
|
|
item.reqAt = time.Now()
|
|
item.peer = peer
|
|
item.requested++
|
|
|
|
dist[peer] = append(dist[peer], hash)
|
|
}
|
|
}
|
|
}
|
|
|
|
for peer, hashes := range dist {
|
|
peer.FetchBlocks(hashes)
|
|
}
|
|
|
|
if len(dist) > 0 {
|
|
self.downloadStartedAt = time.Now()
|
|
}
|
|
}
|
|
|
|
func (self *BlockPool) Start() {
|
|
go self.downloadThread()
|
|
go self.chainThread()
|
|
}
|
|
|
|
func (self *BlockPool) Stop() {
|
|
close(self.quit)
|
|
}
|
|
|
|
func (self *BlockPool) downloadThread() {
|
|
serviceTimer := time.NewTicker(100 * time.Millisecond)
|
|
out:
|
|
for {
|
|
select {
|
|
case <-self.quit:
|
|
break out
|
|
case <-serviceTimer.C:
|
|
// Check if we're catching up. If not distribute the hashes to
|
|
// the peers and download the blockchain
|
|
self.fetchingHashes = false
|
|
eachPeer(self.eth.peers, func(p *Peer, v *list.Element) {
|
|
if p.statusKnown && p.FetchingHashes() {
|
|
self.fetchingHashes = true
|
|
}
|
|
})
|
|
|
|
if len(self.hashes) > 0 {
|
|
self.DistributeHashes()
|
|
}
|
|
|
|
if self.ChainLength < len(self.hashes) {
|
|
self.ChainLength = len(self.hashes)
|
|
}
|
|
|
|
/*
|
|
if !self.fetchingHashes {
|
|
blocks := self.Blocks()
|
|
chain.BlockBy(chain.Number).Sort(blocks)
|
|
|
|
if len(blocks) > 0 {
|
|
if !self.eth.ChainManager().HasBlock(b.PrevHash) && self.pool[string(b.PrevHash)] == nil && !self.fetchingHashes {
|
|
}
|
|
}
|
|
}
|
|
*/
|
|
}
|
|
}
|
|
}
|
|
|
|
func (self *BlockPool) chainThread() {
|
|
procTimer := time.NewTicker(500 * time.Millisecond)
|
|
out:
|
|
for {
|
|
select {
|
|
case <-self.quit:
|
|
break out
|
|
case <-procTimer.C:
|
|
blocks := self.Blocks()
|
|
types.BlockBy(types.Number).Sort(blocks)
|
|
|
|
// Find common block
|
|
for i, block := range blocks {
|
|
if self.eth.ChainManager().HasBlock(block.PrevHash) {
|
|
blocks = blocks[i:]
|
|
break
|
|
}
|
|
}
|
|
|
|
if len(blocks) > 0 {
|
|
if self.eth.ChainManager().HasBlock(blocks[0].PrevHash) {
|
|
for i, block := range blocks[1:] {
|
|
// NOTE: The Ith element in this loop refers to the previous block in
|
|
// outer "blocks"
|
|
if bytes.Compare(block.PrevHash, blocks[i].Hash()) != 0 {
|
|
blocks = blocks[:i]
|
|
|
|
break
|
|
}
|
|
}
|
|
} else {
|
|
blocks = nil
|
|
}
|
|
}
|
|
|
|
if len(blocks) > 0 {
|
|
chainManager := self.eth.ChainManager()
|
|
// Test and import
|
|
bchain := chain.NewChain(blocks)
|
|
_, err := chainManager.TestChain(bchain)
|
|
if err != nil && !chain.IsTDError(err) {
|
|
poollogger.Debugln(err)
|
|
|
|
self.Reset()
|
|
|
|
if self.peer != nil && self.peer.conn != nil {
|
|
poollogger.Debugf("Punishing peer for supplying bad chain (%v)\n", self.peer.conn.RemoteAddr())
|
|
}
|
|
|
|
// This peer gave us bad hashes and made us fetch a bad chain, therefor he shall be punished.
|
|
self.eth.BlacklistPeer(self.peer)
|
|
self.peer.StopWithReason(DiscBadPeer)
|
|
self.td = ethutil.Big0
|
|
self.peer = nil
|
|
} else {
|
|
if !chain.IsTDError(err) {
|
|
chainManager.InsertChain(bchain, func(block *types.Block, messages state.Messages) {
|
|
self.eth.EventMux().Post(chain.NewBlockEvent{block})
|
|
self.eth.EventMux().Post(messages)
|
|
|
|
self.Remove(block.Hash())
|
|
})
|
|
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|