kick of bad peers

This commit is contained in:
obscuren 2014-10-07 11:18:46 +02:00
parent 677836cbee
commit 0015ce1e35
5 changed files with 55 additions and 78 deletions

View File

@ -3,6 +3,7 @@ package eth
import ( import (
"bytes" "bytes"
"container/list" "container/list"
"fmt"
"math" "math"
"math/big" "math/big"
"sync" "sync"
@ -29,7 +30,7 @@ type BlockPool struct {
eth *Ethereum eth *Ethereum
hashPool [][]byte hashes [][]byte
pool map[string]*block pool map[string]*block
td *big.Int td *big.Int
@ -53,12 +54,12 @@ func NewBlockPool(eth *Ethereum) *BlockPool {
} }
func (self *BlockPool) Len() int { func (self *BlockPool) Len() int {
return len(self.hashPool) return len(self.hashes)
} }
func (self *BlockPool) Reset() { func (self *BlockPool) Reset() {
self.pool = make(map[string]*block) self.pool = make(map[string]*block)
self.hashPool = nil self.hashes = nil
} }
func (self *BlockPool) HasLatestHash() bool { func (self *BlockPool) HasLatestHash() bool {
@ -88,6 +89,10 @@ func (self *BlockPool) FetchHashes(peer *Peer) bool {
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 == nil && peer.td.Cmp(highestTd) >= 0) || (self.peer != nil && peer.td.Cmp(self.peer.td) >= 0) || self.peer == peer {
if self.peer != peer { if self.peer != peer {
poollogger.Debugf("Found better suitable peer (%v vs %v)\n", self.td, peer.td) 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.peer = peer
@ -114,7 +119,7 @@ func (self *BlockPool) AddHash(hash []byte, peer *Peer) {
if self.pool[string(hash)] == nil { if self.pool[string(hash)] == nil {
self.pool[string(hash)] = &block{peer, nil, nil, time.Now(), 0} self.pool[string(hash)] = &block{peer, nil, nil, time.Now(), 0}
self.hashPool = append([][]byte{hash}, self.hashPool...) self.hashes = append([][]byte{hash}, self.hashes...)
} }
} }
@ -127,9 +132,12 @@ func (self *BlockPool) Add(b *ethchain.Block, peer *Peer) {
if self.pool[hash] == nil && !self.eth.BlockChain().HasBlock(b.Hash()) { if self.pool[hash] == nil && !self.eth.BlockChain().HasBlock(b.Hash()) {
poollogger.Infof("Got unrequested block (%x...)\n", hash[0:4]) poollogger.Infof("Got unrequested block (%x...)\n", hash[0:4])
self.hashPool = append(self.hashPool, b.Hash()) self.hashes = append(self.hashes, b.Hash())
self.pool[hash] = &block{peer, peer, b, time.Now(), 0} self.pool[hash] = &block{peer, peer, b, time.Now(), 0}
fmt.Println("1.", !self.eth.BlockChain().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.BlockChain().HasBlock(b.PrevHash) && self.pool[string(b.PrevHash)] == nil && !self.fetchingHashes { if !self.eth.BlockChain().HasBlock(b.PrevHash) && self.pool[string(b.PrevHash)] == nil && !self.fetchingHashes {
poollogger.Infof("Unknown chain, requesting (%x...)\n", b.PrevHash[0:4]) poollogger.Infof("Unknown chain, requesting (%x...)\n", b.PrevHash[0:4])
peer.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlockHashesTy, []interface{}{b.Hash(), uint32(256)})) peer.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlockHashesTy, []interface{}{b.Hash(), uint32(256)}))
@ -145,28 +153,10 @@ func (self *BlockPool) Remove(hash []byte) {
self.mut.Lock() self.mut.Lock()
defer self.mut.Unlock() defer self.mut.Unlock()
self.hashPool = ethutil.DeleteFromByteSlice(self.hashPool, hash) self.hashes = ethutil.DeleteFromByteSlice(self.hashes, hash)
delete(self.pool, string(hash)) delete(self.pool, string(hash))
} }
func (self *BlockPool) ProcessCanonical(f func(block *ethchain.Block)) (procAmount int) {
blocks := self.Blocks()
ethchain.BlockBy(ethchain.Number).Sort(blocks)
for _, block := range blocks {
if self.eth.BlockChain().HasBlock(block.PrevHash) {
procAmount++
f(block)
self.Remove(block.Hash())
}
}
return
}
func (self *BlockPool) DistributeHashes() { func (self *BlockPool) DistributeHashes() {
self.mut.Lock() self.mut.Lock()
defer self.mut.Unlock() defer self.mut.Unlock()
@ -178,8 +168,8 @@ func (self *BlockPool) DistributeHashes() {
) )
num := int(math.Min(float64(amount), float64(len(self.pool)))) num := int(math.Min(float64(amount), float64(len(self.pool))))
for i, j := 0, 0; i < len(self.hashPool) && j < num; i++ { for i, j := 0, 0; i < len(self.hashes) && j < num; i++ {
hash := self.hashPool[i] hash := self.hashes[i]
item := self.pool[string(hash)] item := self.pool[string(hash)]
if item != nil && item.block == nil { if item != nil && item.block == nil {
@ -193,7 +183,7 @@ func (self *BlockPool) DistributeHashes() {
peer = item.from peer = item.from
} else { } else {
// Remove it // Remove it
self.hashPool = ethutil.DeleteFromByteSlice(self.hashPool, hash) self.hashes = ethutil.DeleteFromByteSlice(self.hashes, hash)
delete(self.pool, string(hash)) delete(self.pool, string(hash))
} }
} else if lastFetchFailed || item.peer == nil { } else if lastFetchFailed || item.peer == nil {
@ -250,17 +240,31 @@ out:
} }
}) })
if len(self.hashes) > 0 {
self.DistributeHashes() self.DistributeHashes()
if self.ChainLength < len(self.hashPool) {
self.ChainLength = len(self.hashPool)
} }
if self.ChainLength < len(self.hashes) {
self.ChainLength = len(self.hashes)
}
/*
if !self.fetchingHashes {
blocks := self.Blocks()
ethchain.BlockBy(ethchain.Number).Sort(blocks)
if len(blocks) > 0 {
if !self.eth.BlockChain().HasBlock(b.PrevHash) && self.pool[string(b.PrevHash)] == nil && !self.fetchingHashes {
}
}
}
*/
} }
} }
} }
func (self *BlockPool) chainThread() { func (self *BlockPool) chainThread() {
procTimer := time.NewTicker(1000 * time.Millisecond) procTimer := time.NewTicker(500 * time.Millisecond)
out: out:
for { for {
select { select {
@ -294,14 +298,8 @@ out:
} }
} }
if len(blocks) > 0 {
self.eth.Eventer().Post("blocks", blocks)
}
var err error var err error
for i, block := range blocks { for i, block := range blocks {
//self.eth.Eventer().Post("block", block)
err = self.eth.StateManager().Process(block, false) err = self.eth.StateManager().Process(block, false)
if err != nil { if err != nil {
poollogger.Infoln(err) poollogger.Infoln(err)
@ -318,10 +316,6 @@ out:
if err != nil { if err != nil {
self.Reset() self.Reset()
// Remove this bad chain
for _, block := range blocks {
self.Remove(block.Hash())
}
poollogger.Debugf("Punishing peer for supplying bad chain (%v)\n", self.peer.conn.RemoteAddr()) 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. // This peer gave us bad hashes and made us fetch a bad chain, therefor he shall be punished.
@ -330,16 +324,6 @@ out:
self.td = ethutil.Big0 self.td = ethutil.Big0
self.peer = nil self.peer = nil
} }
/*
// Handle in batches of 4k
//max := int(math.Min(4000, float64(len(blocks))))
for _, block := range blocks {
self.eth.Eventer().Post("block", block)
self.Remove(block.Hash())
}
*/
} }
} }
} }

View File

@ -216,7 +216,7 @@ func (self *Filter) bloomFilter(block *Block) bool {
fk := append([]byte("bloom"), block.Hash()...) fk := append([]byte("bloom"), block.Hash()...)
bin, err := self.eth.Db().Get(fk) bin, err := self.eth.Db().Get(fk)
if err != nil { if err != nil {
panic(err) fmt.Println(err)
} }
bloom := NewBloomFilter(bin) bloom := NewBloomFilter(bin)

View File

@ -33,9 +33,9 @@ func NewJSBlock(block *ethchain.Block) *JSBlock {
return &JSBlock{} return &JSBlock{}
} }
var ptxs []JSTransaction var ptxs []*JSTransaction
for _, tx := range block.Transactions() { for _, tx := range block.Transactions() {
ptxs = append(ptxs, *NewJSTx(tx, block.State())) ptxs = append(ptxs, NewJSTx(tx, block.State()))
} }
list := ethutil.NewList(ptxs) list := ethutil.NewList(ptxs)

View File

@ -3,12 +3,14 @@ package ethutil
import ( import (
"encoding/json" "encoding/json"
"reflect" "reflect"
"sync"
) )
// The list type is an anonymous slice handler which can be used // The list type is an anonymous slice handler which can be used
// for containing any slice type to use in an environment which // for containing any slice type to use in an environment which
// does not support slice types (e.g., JavaScript, QML) // does not support slice types (e.g., JavaScript, QML)
type List struct { type List struct {
mut sync.Mutex
val interface{} val interface{}
list reflect.Value list reflect.Value
Length int Length int
@ -21,7 +23,7 @@ func NewList(t interface{}) *List {
panic("list container initialized with a non-slice type") panic("list container initialized with a non-slice type")
} }
return &List{t, list, list.Len()} return &List{sync.Mutex{}, t, list, list.Len()}
} }
func EmptyList() *List { func EmptyList() *List {
@ -30,8 +32,10 @@ func EmptyList() *List {
// Get N element from the embedded slice. Returns nil if OOB. // Get N element from the embedded slice. Returns nil if OOB.
func (self *List) Get(i int) interface{} { func (self *List) Get(i int) interface{} {
if self.list.Len() > i { if self.list.Len() > i {
self.mut.Lock()
defer self.mut.Unlock()
i := self.list.Index(i).Interface() i := self.list.Index(i).Interface()
return i return i
@ -51,6 +55,9 @@ func (self *List) GetAsJson(i int) interface{} {
// Appends value at the end of the slice. Panics when incompatible value // Appends value at the end of the slice. Panics when incompatible value
// is given. // is given.
func (self *List) Append(v interface{}) { func (self *List) Append(v interface{}) {
self.mut.Lock()
defer self.mut.Unlock()
self.list = reflect.Append(self.list, reflect.ValueOf(v)) self.list = reflect.Append(self.list, reflect.ValueOf(v))
self.Length = self.list.Len() self.Length = self.list.Len()
} }

26
peer.go
View File

@ -516,10 +516,11 @@ func (p *Peer) HandleInbound() {
blockPool.AddHash(hash, p) blockPool.AddHash(hash, p)
} }
if !foundCommonHash && msg.Data.Len() != 0 { if !foundCommonHash {
if !p.FetchHashes() { //if !p.FetchHashes() {
p.doneFetchingHashes = true // p.doneFetchingHashes = true
} //}
p.FetchHashes()
} else { } else {
peerlogger.Infof("Found common hash (%x...)\n", p.lastReceivedHash[0:4]) peerlogger.Infof("Found common hash (%x...)\n", p.lastReceivedHash[0:4])
p.doneFetchingHashes = true p.doneFetchingHashes = true
@ -533,8 +534,6 @@ func (p *Peer) HandleInbound() {
it := msg.Data.NewIterator() it := msg.Data.NewIterator()
for it.Next() { for it.Next() {
block := ethchain.NewBlockFromRlpValue(it.Value()) block := ethchain.NewBlockFromRlpValue(it.Value())
//fmt.Printf("%v %x - %x\n", block.Number, block.Hash()[0:4], block.PrevHash[0:4])
blockPool.Add(block, p) blockPool.Add(block, p)
p.lastBlockReceived = time.Now() p.lastBlockReceived = time.Now()
@ -557,21 +556,8 @@ func (self *Peer) FetchBlocks(hashes [][]byte) {
func (self *Peer) FetchHashes() bool { func (self *Peer) FetchHashes() bool {
blockPool := self.ethereum.blockPool blockPool := self.ethereum.blockPool
return blockPool.FetchHashes(self) return blockPool.FetchHashes(self)
/*
if self.td.Cmp(self.ethereum.HighestTDPeer()) >= 0 {
blockPool.td = self.td
if !blockPool.HasLatestHash() {
self.doneFetchingHashes = false
const amount = 256
peerlogger.Debugf("Fetching hashes (%d)\n", amount)
self.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlockHashesTy, []interface{}{self.lastReceivedHash, uint32(amount)}))
}
}
*/
} }
func (self *Peer) FetchingHashes() bool { func (self *Peer) FetchingHashes() bool {