plugeth/block_pool.go

342 lines
7.9 KiB
Go
Raw Normal View History

package eth
import (
"bytes"
2014-09-24 09:41:57 +00:00
"container/list"
2014-10-07 09:18:46 +00:00
"fmt"
"math"
"math/big"
"sync"
2014-09-24 09:41:57 +00:00
"time"
2014-10-31 09:59:17 +00:00
"github.com/ethereum/go-ethereum/chain"
"github.com/ethereum/go-ethereum/ethutil"
2014-10-31 11:56:05 +00:00
"github.com/ethereum/go-ethereum/logger"
2014-10-31 13:53:42 +00:00
"github.com/ethereum/go-ethereum/wire"
)
2014-10-31 11:56:05 +00:00
var poollogger = logger.NewLogger("BPOOL")
type block struct {
from *Peer
2014-09-24 09:41:57 +00:00
peer *Peer
2014-10-31 09:59:17 +00:00
block *chain.Block
2014-09-24 09:41:57 +00:00
reqAt time.Time
requested int
}
type BlockPool struct {
mut sync.Mutex
eth *Ethereum
2014-10-07 09:18:46 +00:00
hashes [][]byte
pool map[string]*block
2014-09-24 09:41:57 +00:00
td *big.Int
quit chan bool
2014-09-28 12:52:58 +00:00
fetchingHashes bool
downloadStartedAt time.Time
ChainLength, BlocksProcessed int
2014-10-02 15:03:48 +00:00
peer *Peer
}
func NewBlockPool(eth *Ethereum) *BlockPool {
return &BlockPool{
eth: eth,
pool: make(map[string]*block),
td: ethutil.Big0,
2014-09-24 09:41:57 +00:00
quit: make(chan bool),
}
}
2014-09-17 13:58:02 +00:00
func (self *BlockPool) Len() int {
2014-10-07 09:18:46 +00:00
return len(self.hashes)
2014-09-17 13:58:02 +00:00
}
func (self *BlockPool) Reset() {
self.pool = make(map[string]*block)
2014-10-07 09:18:46 +00:00
self.hashes = nil
}
func (self *BlockPool) HasLatestHash() bool {
2014-09-28 12:52:58 +00:00
self.mut.Lock()
defer self.mut.Unlock()
2014-10-20 09:53:11 +00:00
return self.pool[string(self.eth.ChainManager().CurrentBlock.Hash())] != nil
}
func (self *BlockPool) HasCommonHash(hash []byte) bool {
2014-10-20 09:53:11 +00:00
return self.eth.ChainManager().GetBlock(hash) != nil
}
2014-10-31 09:59:17 +00:00
func (self *BlockPool) Blocks() (blocks chain.Blocks) {
2014-09-28 12:52:58 +00:00
for _, item := range self.pool {
if item.block != nil {
blocks = append(blocks, item.block)
}
}
return
}
func (self *BlockPool) FetchHashes(peer *Peer) bool {
2014-10-02 15:03:48 +00:00
highestTd := self.eth.HighestTDPeer()
2014-10-08 09:59:44 +00:00
if (self.peer == nil && peer.td.Cmp(highestTd) >= 0) || (self.peer != nil && peer.td.Cmp(self.peer.td) > 0) || self.peer == peer {
2014-10-02 15:03:48 +00:00
if self.peer != peer {
poollogger.Debugf("Found better suitable peer (%v vs %v)\n", self.td, peer.td)
2014-10-07 09:18:46 +00:00
if self.peer != nil {
self.peer.doneFetchingHashes = true
}
2014-10-02 15:03:48 +00:00
}
self.peer = peer
self.td = peer.td
if !self.HasLatestHash() {
peer.doneFetchingHashes = false
const amount = 256
2014-10-08 09:59:44 +00:00
peerlogger.Debugf("Fetching hashes (%d) %x...\n", amount, peer.lastReceivedHash[0:4])
2014-10-31 13:53:42 +00:00
peer.QueueMessage(wire.NewMessage(wire.MsgGetBlockHashesTy, []interface{}{peer.lastReceivedHash, uint32(amount)}))
2014-10-02 15:03:48 +00:00
}
return true
2014-10-02 15:03:48 +00:00
}
return false
2014-10-02 15:03:48 +00:00
}
func (self *BlockPool) AddHash(hash []byte, peer *Peer) {
2014-09-28 12:52:58 +00:00
self.mut.Lock()
defer self.mut.Unlock()
if self.pool[string(hash)] == nil {
self.pool[string(hash)] = &block{peer, nil, nil, time.Now(), 0}
2014-10-07 09:18:46 +00:00
self.hashes = append([][]byte{hash}, self.hashes...)
}
}
2014-10-31 09:59:17 +00:00
func (self *BlockPool) Add(b *chain.Block, peer *Peer) {
2014-10-08 10:29:49 +00:00
self.addBlock(b, peer, false)
}
2014-10-31 09:59:17 +00:00
func (self *BlockPool) AddNew(b *chain.Block, peer *Peer) {
2014-10-08 10:29:49 +00:00
self.addBlock(b, peer, true)
}
2014-10-31 09:59:17 +00:00
func (self *BlockPool) addBlock(b *chain.Block, peer *Peer, newBlock bool) {
2014-09-28 12:52:58 +00:00
self.mut.Lock()
defer self.mut.Unlock()
hash := string(b.Hash())
2014-10-20 09:53:11 +00:00
if self.pool[hash] == nil && !self.eth.ChainManager().HasBlock(b.Hash()) {
2014-09-26 18:19:11 +00:00
poollogger.Infof("Got unrequested block (%x...)\n", hash[0:4])
2014-10-07 09:18:46 +00:00
self.hashes = append(self.hashes, b.Hash())
self.pool[hash] = &block{peer, peer, b, time.Now(), 0}
2014-09-25 14:57:49 +00:00
2014-10-08 10:29:49 +00:00
// The following is only performed on an unrequested new block
if newBlock {
2014-10-20 09:53:11 +00:00
fmt.Println("1.", !self.eth.ChainManager().HasBlock(b.PrevHash), ethutil.Bytes2Hex(b.Hash()[0:4]), ethutil.Bytes2Hex(b.PrevHash[0:4]))
2014-10-08 10:29:49 +00:00
fmt.Println("2.", self.pool[string(b.PrevHash)] == nil)
fmt.Println("3.", !self.fetchingHashes)
2014-10-20 09:53:11 +00:00
if !self.eth.ChainManager().HasBlock(b.PrevHash) && self.pool[string(b.PrevHash)] == nil && !self.fetchingHashes {
2014-10-08 10:29:49 +00:00
poollogger.Infof("Unknown chain, requesting (%x...)\n", b.PrevHash[0:4])
2014-10-31 13:53:42 +00:00
peer.QueueMessage(wire.NewMessage(wire.MsgGetBlockHashesTy, []interface{}{b.Hash(), uint32(256)}))
2014-10-08 10:29:49 +00:00
}
2014-09-25 14:57:49 +00:00
}
2014-09-15 13:42:12 +00:00
} else if self.pool[hash] != nil {
self.pool[hash].block = b
}
self.BlocksProcessed++
}
2014-09-28 12:52:58 +00:00
func (self *BlockPool) Remove(hash []byte) {
self.mut.Lock()
defer self.mut.Unlock()
2014-10-07 09:18:46 +00:00
self.hashes = ethutil.DeleteFromByteSlice(self.hashes, hash)
2014-09-28 12:52:58 +00:00
delete(self.pool, string(hash))
}
func (self *BlockPool) DistributeHashes() {
2014-09-28 12:52:58 +00:00
self.mut.Lock()
defer self.mut.Unlock()
var (
peerLen = self.eth.peers.Len()
2014-09-28 12:52:58 +00:00
amount = 256 * peerLen
dist = make(map[*Peer][][]byte)
)
num := int(math.Min(float64(amount), float64(len(self.pool))))
2014-10-07 09:18:46 +00:00
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
2014-09-28 12:52:58 +00:00
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
2014-10-07 09:18:46 +00:00
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 {
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)
}
2014-09-28 12:52:58 +00:00
if len(dist) > 0 {
self.downloadStartedAt = time.Now()
}
}
2014-09-24 09:41:57 +00:00
func (self *BlockPool) Start() {
2014-09-28 12:52:58 +00:00
go self.downloadThread()
go self.chainThread()
2014-09-24 09:41:57 +00:00
}
func (self *BlockPool) Stop() {
close(self.quit)
}
2014-09-28 12:52:58 +00:00
func (self *BlockPool) downloadThread() {
2014-09-24 09:41:57 +00:00
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
2014-09-28 12:52:58 +00:00
self.fetchingHashes = false
2014-09-24 09:41:57 +00:00
eachPeer(self.eth.peers, func(p *Peer, v *list.Element) {
if p.statusKnown && p.FetchingHashes() {
2014-09-28 12:52:58 +00:00
self.fetchingHashes = true
2014-09-24 09:41:57 +00:00
}
})
2014-10-07 09:18:46 +00:00
if len(self.hashes) > 0 {
self.DistributeHashes()
}
2014-10-07 09:18:46 +00:00
if self.ChainLength < len(self.hashes) {
self.ChainLength = len(self.hashes)
2014-09-24 09:41:57 +00:00
}
2014-10-07 09:18:46 +00:00
/*
if !self.fetchingHashes {
blocks := self.Blocks()
2014-10-31 09:59:17 +00:00
chain.BlockBy(chain.Number).Sort(blocks)
2014-10-07 09:18:46 +00:00
if len(blocks) > 0 {
2014-10-20 09:53:11 +00:00
if !self.eth.ChainManager().HasBlock(b.PrevHash) && self.pool[string(b.PrevHash)] == nil && !self.fetchingHashes {
2014-10-07 09:18:46 +00:00
}
}
}
*/
2014-09-28 12:52:58 +00:00
}
}
}
func (self *BlockPool) chainThread() {
2014-10-07 09:18:46 +00:00
procTimer := time.NewTicker(500 * time.Millisecond)
2014-09-28 12:52:58 +00:00
out:
for {
select {
case <-self.quit:
break out
2014-09-24 09:41:57 +00:00
case <-procTimer.C:
blocks := self.Blocks()
2014-10-31 09:59:17 +00:00
chain.BlockBy(chain.Number).Sort(blocks)
2014-10-02 15:03:48 +00:00
// Find common block
for i, block := range blocks {
2014-10-20 09:53:11 +00:00
if self.eth.ChainManager().HasBlock(block.PrevHash) {
2014-10-02 15:03:48 +00:00
blocks = blocks[i:]
break
}
}
if len(blocks) > 0 {
2014-10-20 09:53:11 +00:00
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
}
}
2014-10-29 13:20:42 +00:00
// TODO figure out whether we were catching up
// If caught up and just a new block has been propagated:
// sm.eth.EventMux().Post(NewBlockEvent{block})
// otherwise process and don't emit anything
if len(blocks) > 0 {
chainManager := self.eth.ChainManager()
// Test and import
2014-11-11 11:16:36 +00:00
bchain := chain.NewChain(blocks)
_, err := chainManager.TestChain(bchain)
if err != nil && !chain.IsTDError(err) {
poollogger.Debugln(err)
self.Reset()
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 {
2014-11-11 11:16:36 +00:00
chainManager.InsertChain(bchain)
for _, block := range blocks {
self.Remove(block.Hash())
}
2014-10-02 15:03:48 +00:00
}
}
2014-09-24 09:41:57 +00:00
}
}
}