From 0015ce1e353f52cca818d11f566b9a656fb85f24 Mon Sep 17 00:00:00 2001 From: obscuren Date: Tue, 7 Oct 2014 11:18:46 +0200 Subject: kick of bad peers --- block_pool.go | 88 ++++++++++++++++++++++------------------------------- ethchain/filter.go | 2 +- ethpipe/js_types.go | 4 +-- ethutil/list.go | 11 +++++-- peer.go | 26 ++++------------ 5 files changed, 54 insertions(+), 77 deletions(-) diff --git a/block_pool.go b/block_pool.go index a951672c3..15d8ba3b9 100644 --- a/block_pool.go +++ b/block_pool.go @@ -3,6 +3,7 @@ package eth import ( "bytes" "container/list" + "fmt" "math" "math/big" "sync" @@ -29,8 +30,8 @@ type BlockPool struct { eth *Ethereum - hashPool [][]byte - pool map[string]*block + hashes [][]byte + pool map[string]*block td *big.Int quit chan bool @@ -53,12 +54,12 @@ func NewBlockPool(eth *Ethereum) *BlockPool { } func (self *BlockPool) Len() int { - return len(self.hashPool) + return len(self.hashes) } func (self *BlockPool) Reset() { self.pool = make(map[string]*block) - self.hashPool = nil + self.hashes = nil } 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 != 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 @@ -114,7 +119,7 @@ func (self *BlockPool) AddHash(hash []byte, peer *Peer) { if self.pool[string(hash)] == nil { 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()) { 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} + 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 { poollogger.Infof("Unknown chain, requesting (%x...)\n", b.PrevHash[0:4]) peer.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlockHashesTy, []interface{}{b.Hash(), uint32(256)})) @@ -145,28 +153,10 @@ func (self *BlockPool) Remove(hash []byte) { self.mut.Lock() defer self.mut.Unlock() - self.hashPool = ethutil.DeleteFromByteSlice(self.hashPool, hash) + self.hashes = ethutil.DeleteFromByteSlice(self.hashes, 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() { self.mut.Lock() defer self.mut.Unlock() @@ -178,8 +168,8 @@ func (self *BlockPool) DistributeHashes() { ) num := int(math.Min(float64(amount), float64(len(self.pool)))) - for i, j := 0, 0; i < len(self.hashPool) && j < num; i++ { - hash := self.hashPool[i] + 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 { @@ -193,7 +183,7 @@ func (self *BlockPool) DistributeHashes() { peer = item.from } else { // Remove it - self.hashPool = ethutil.DeleteFromByteSlice(self.hashPool, hash) + self.hashes = ethutil.DeleteFromByteSlice(self.hashes, hash) delete(self.pool, string(hash)) } } else if lastFetchFailed || item.peer == nil { @@ -250,17 +240,31 @@ out: } }) - self.DistributeHashes() + if len(self.hashes) > 0 { + 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() { - procTimer := time.NewTicker(1000 * time.Millisecond) + procTimer := time.NewTicker(500 * time.Millisecond) out: for { select { @@ -294,14 +298,8 @@ out: } } - if len(blocks) > 0 { - self.eth.Eventer().Post("blocks", blocks) - - } - var err error for i, block := range blocks { - //self.eth.Eventer().Post("block", block) err = self.eth.StateManager().Process(block, false) if err != nil { poollogger.Infoln(err) @@ -318,10 +316,6 @@ out: if err != nil { 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()) // 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.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()) - } - */ } } } diff --git a/ethchain/filter.go b/ethchain/filter.go index 0a97b3f98..447163b68 100644 --- a/ethchain/filter.go +++ b/ethchain/filter.go @@ -216,7 +216,7 @@ func (self *Filter) bloomFilter(block *Block) bool { fk := append([]byte("bloom"), block.Hash()...) bin, err := self.eth.Db().Get(fk) if err != nil { - panic(err) + fmt.Println(err) } bloom := NewBloomFilter(bin) diff --git a/ethpipe/js_types.go b/ethpipe/js_types.go index ca85da85a..b8b546980 100644 --- a/ethpipe/js_types.go +++ b/ethpipe/js_types.go @@ -33,9 +33,9 @@ func NewJSBlock(block *ethchain.Block) *JSBlock { return &JSBlock{} } - var ptxs []JSTransaction + var ptxs []*JSTransaction for _, tx := range block.Transactions() { - ptxs = append(ptxs, *NewJSTx(tx, block.State())) + ptxs = append(ptxs, NewJSTx(tx, block.State())) } list := ethutil.NewList(ptxs) diff --git a/ethutil/list.go b/ethutil/list.go index 4fb36224f..6919a02f5 100644 --- a/ethutil/list.go +++ b/ethutil/list.go @@ -3,12 +3,14 @@ package ethutil import ( "encoding/json" "reflect" + "sync" ) // The list type is an anonymous slice handler which can be used // for containing any slice type to use in an environment which // does not support slice types (e.g., JavaScript, QML) type List struct { + mut sync.Mutex val interface{} list reflect.Value Length int @@ -21,7 +23,7 @@ func NewList(t interface{}) *List { 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 { @@ -30,8 +32,10 @@ func EmptyList() *List { // Get N element from the embedded slice. Returns nil if OOB. func (self *List) Get(i int) interface{} { - if self.list.Len() > i { + self.mut.Lock() + defer self.mut.Unlock() + i := self.list.Index(i).Interface() 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 // is given. func (self *List) Append(v interface{}) { + self.mut.Lock() + defer self.mut.Unlock() + self.list = reflect.Append(self.list, reflect.ValueOf(v)) self.Length = self.list.Len() } diff --git a/peer.go b/peer.go index 763658dd5..63def37ed 100644 --- a/peer.go +++ b/peer.go @@ -516,10 +516,11 @@ func (p *Peer) HandleInbound() { blockPool.AddHash(hash, p) } - if !foundCommonHash && msg.Data.Len() != 0 { - if !p.FetchHashes() { - p.doneFetchingHashes = true - } + if !foundCommonHash { + //if !p.FetchHashes() { + // p.doneFetchingHashes = true + //} + p.FetchHashes() } else { peerlogger.Infof("Found common hash (%x...)\n", p.lastReceivedHash[0:4]) p.doneFetchingHashes = true @@ -533,8 +534,6 @@ func (p *Peer) HandleInbound() { it := msg.Data.NewIterator() for it.Next() { 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) p.lastBlockReceived = time.Now() @@ -557,21 +556,8 @@ func (self *Peer) FetchBlocks(hashes [][]byte) { func (self *Peer) FetchHashes() bool { blockPool := self.ethereum.blockPool - 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)})) - } - } - */ + return blockPool.FetchHashes(self) } func (self *Peer) FetchingHashes() bool { -- cgit v1.2.3