Merge branch 'tmp' into develop

Conflicts:
	peer.go

7 files changed, 169 insertions, 66 deletions

diff --git a/block_pool.go b/block_pool.go
index 6753308b6..88d1c3739 100644
--- a/block_pool.go
+++ b/block_pool.go
@@ -1,17 +1,27 @@
 package eth
 
 import (
+	"bytes"
+	"container/list"
 	"math"
 	"math/big"
 	"sync"
+	"time"
 
 	"github.com/ethereum/eth-go/ethchain"
+	"github.com/ethereum/eth-go/ethlog"
 	"github.com/ethereum/eth-go/ethutil"
+	"github.com/ethereum/eth-go/ethwire"
 )
 
+var poollogger = ethlog.NewLogger("[BPOOL]")
+
 type block struct {
-	peer  *Peer
-	block *ethchain.Block
+	from      *Peer
+	peer      *Peer
+	block     *ethchain.Block
+	reqAt     time.Time
+	requested int
 }
 
 type BlockPool struct {
@@ -22,7 +32,10 @@ type BlockPool struct {
 	hashPool [][]byte
 	pool     map[string]*block
 
-	td *big.Int
+	td   *big.Int
+	quit chan bool
+
+	ChainLength, BlocksProcessed int
 }
 
 func NewBlockPool(eth *Ethereum) *BlockPool {
@@ -30,6 +43,7 @@ func NewBlockPool(eth *Ethereum) *BlockPool {
 		eth:  eth,
 		pool: make(map[string]*block),
 		td:   ethutil.Big0,
+		quit: make(chan bool),
 	}
 }
 
@@ -45,9 +59,9 @@ func (self *BlockPool) HasCommonHash(hash []byte) bool {
 	return self.eth.BlockChain().GetBlock(hash) != nil
 }
 
-func (self *BlockPool) AddHash(hash []byte) {
+func (self *BlockPool) AddHash(hash []byte, peer *Peer) {
 	if self.pool[string(hash)] == nil {
-		self.pool[string(hash)] = &block{nil, nil}
+		self.pool[string(hash)] = &block{peer, nil, nil, time.Now(), 0}
 
 		self.hashPool = append([][]byte{hash}, self.hashPool...)
 	}
@@ -58,24 +72,54 @@ func (self *BlockPool) SetBlock(b *ethchain.Block, peer *Peer) {
 
 	if self.pool[hash] == nil && !self.eth.BlockChain().HasBlock(b.Hash()) {
 		self.hashPool = append(self.hashPool, b.Hash())
-		self.pool[hash] = &block{peer, b}
+		self.pool[hash] = &block{peer, peer, b, time.Now(), 0}
 	} else if self.pool[hash] != nil {
 		self.pool[hash].block = b
 	}
+
+	self.BlocksProcessed++
 }
 
-func (self *BlockPool) CheckLinkAndProcess(f func(block *ethchain.Block)) {
+func (self *BlockPool) getParent(block *ethchain.Block) *ethchain.Block {
+	for _, item := range self.pool {
+		if item.block != nil {
+			if bytes.Compare(item.block.Hash(), block.PrevHash) == 0 {
+				return item.block
+			}
+		}
+	}
+
+	return nil
+}
 
+func (self *BlockPool) GetChainFromBlock(block *ethchain.Block) ethchain.Blocks {
 	var blocks ethchain.Blocks
+
+	for b := block; b != nil; b = self.getParent(b) {
+		blocks = append(ethchain.Blocks{b}, blocks...)
+	}
+
+	return blocks
+}
+
+func (self *BlockPool) Blocks() (blocks ethchain.Blocks) {
 	for _, item := range self.pool {
 		if item.block != nil {
 			blocks = append(blocks, item.block)
 		}
 	}
 
+	return
+}
+
+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)
 
 			hash := block.Hash()
@@ -84,23 +128,108 @@ func (self *BlockPool) CheckLinkAndProcess(f func(block *ethchain.Block)) {
 		}
 
 	}
+
+	return
 }
 
-func (self *BlockPool) Take(amount int, peer *Peer) (hashes [][]byte) {
-	self.mut.Lock()
-	defer self.mut.Unlock()
+func (self *BlockPool) DistributeHashes() {
+	var (
+		peerLen = self.eth.peers.Len()
+		amount  = 200 * peerLen
+		dist    = make(map[*Peer][][]byte)
+	)
 
 	num := int(math.Min(float64(amount), float64(len(self.pool))))
-	j := 0
-	for i := 0; i < len(self.hashPool) && j < num; i++ {
-		hash := string(self.hashPool[i])
-		if self.pool[hash] != nil && (self.pool[hash].peer == nil || self.pool[hash].peer == peer) && self.pool[hash].block == nil {
-			self.pool[hash].peer = peer
-
-			hashes = append(hashes, self.hashPool[i])
-			j++
+	for i, j := 0, 0; i < len(self.hashPool) && j < num; i++ {
+		hash := self.hashPool[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 > 0 && item.peer != nil {
+				if item.requested < 100 {
+					// Select peer the hash was retrieved off
+					peer = item.from
+				} else {
+					// Remove it
+					self.hashPool = ethutil.DeleteFromByteSlice(self.hashPool, 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)
+			}
 		}
 	}
 
-	return
+	for peer, hashes := range dist {
+		peer.FetchBlocks(hashes)
+	}
+}
+
+func (self *BlockPool) Start() {
+	go self.update()
+}
+
+func (self *BlockPool) Stop() {
+	close(self.quit)
+}
+
+func (self *BlockPool) update() {
+	serviceTimer := time.NewTicker(100 * time.Millisecond)
+	procTimer := time.NewTicker(500 * 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
+			done := true
+			eachPeer(self.eth.peers, func(p *Peer, v *list.Element) {
+				if p.statusKnown && p.FetchingHashes() {
+					done = false
+				}
+			})
+
+			if done && len(self.hashPool) > 0 {
+				self.DistributeHashes()
+			}
+
+			if self.ChainLength < len(self.hashPool) {
+				self.ChainLength = len(self.hashPool)
+			}
+		case <-procTimer.C:
+			// XXX We can optimize this lifting this on to a new goroutine.
+			// We'd need to make sure that the pools are properly protected by a mutex
+			amount := self.ProcessCanonical(func(block *ethchain.Block) {
+				err := self.eth.StateManager().Process(block, false)
+				if err != nil {
+					poollogger.Infoln(err)
+				}
+			})
+
+			// Do not propagate to the network on catchups
+			if amount == 1 {
+				block := self.eth.BlockChain().CurrentBlock
+				self.eth.Broadcast(ethwire.MsgBlockTy, []interface{}{block.Value().Val})
+			}
+		}
+	}
 }
diff --git a/ethchain/state_manager.go b/ethchain/state_manager.go
index b0ea754f4..888c68bb9 100644
--- a/ethchain/state_manager.go
+++ b/ethchain/state_manager.go
@@ -250,15 +250,13 @@ func (sm *StateManager) Process(block *Block, dontReact bool) (err error) {
 		fk := append([]byte("bloom"), block.Hash()...)
 		sm.Ethereum.Db().Put(fk, filter.Bin())
 
-		statelogger.Infof("Added block #%d (%x)\n", block.Number, block.Hash())
+		statelogger.Debugf("Added block #%d (%x)\n", block.Number, block.Hash())
 		if dontReact == false {
 			sm.Ethereum.Reactor().Post("newBlock", block)
 
 			state.Manifest().Reset()
 		}
 
-		sm.Ethereum.Broadcast(ethwire.MsgBlockTy, []interface{}{block.Value().Val})
-
 		sm.Ethereum.TxPool().RemoveInvalid(state)
 	} else {
 		statelogger.Errorln("total diff failed")
diff --git a/ethereum.go b/ethereum.go
index 8f667b786..389b9d4e4 100644
--- a/ethereum.go
+++ b/ethereum.go
@@ -158,6 +158,9 @@ func (s *Ethereum) StateManager() *ethchain.StateManager {
 func (s *Ethereum) TxPool() *ethchain.TxPool {
 	return s.txPool
 }
+func (s *Ethereum) BlockPool() *BlockPool {
+	return s.blockPool
+}
 func (self *Ethereum) Db() ethutil.Database {
 	return self.db
 }
@@ -503,6 +506,7 @@ func (s *Ethereum) Stop() {
 	s.stateManager.Stop()
 	s.reactor.Flush()
 	s.reactor.Stop()
+	s.blockPool.Stop()
 
 	ethlogger.Infoln("Server stopped")
 	close(s.shutdownChan)
diff --git a/ethminer/miner.go b/ethminer/miner.go
index b74f90e0c..299a5204a 100644
--- a/ethminer/miner.go
+++ b/ethminer/miner.go
@@ -3,7 +3,6 @@ package ethminer
 import (
 	"bytes"
 	"sort"
-	"time"
 
 	"github.com/ethereum/eth-go/ethchain"
 	"github.com/ethereum/eth-go/ethlog"
@@ -136,12 +135,6 @@ func (miner *Miner) listener() {
 				}
 			}
 		default:
-			// This hack is only temporarily
-			if len(miner.txs) == 0 {
-				time.Sleep(2 * time.Second)
-				continue
-			}
-
 			miner.mineNewBlock()
 		}
 	}
diff --git a/ethtrie/trie.go b/ethtrie/trie.go
index e6e09dd0d..695ff436a 100644
--- a/ethtrie/trie.go
+++ b/ethtrie/trie.go
@@ -196,8 +196,8 @@ func (t *Trie) Update(key, value string) {
 }
 
 func (t *Trie) Get(key string) string {
-	t.mut.RLock()
-	defer t.mut.RUnlock()
+	t.mut.Lock()
+	defer t.mut.Unlock()
 
 	k := CompactHexDecode(key)
 	c := ethutil.NewValue(t.getState(t.Root, k))
diff --git a/ethvm/vm.go b/ethvm/vm.go
index 7aff320f9..bb14d75da 100644
--- a/ethvm/vm.go
+++ b/ethvm/vm.go
@@ -961,7 +961,6 @@ func (self *Message) Addr() []byte {
 }
 
 func (self *Message) Exec(codeAddr []byte, caller ClosureRef) (ret []byte, err error) {
-	fmt.Printf("%x %x\n", codeAddr[0:4], self.address[0:4])
 	queue := self.vm.queue
 	self.vm.queue = list.New()
 
diff --git a/peer.go b/peer.go
index 00987c123..06df5fab6 100644
--- a/peer.go
+++ b/peer.go
@@ -131,7 +131,7 @@ type Peer struct {
 	// Last received pong message
 	lastPong          int64
 	lastBlockReceived time.Time
-	lastHashReceived  time.Time
+	LastHashReceived  time.Time
 
 	host             []byte
 	port             uint16
@@ -177,6 +177,7 @@ func NewPeer(conn net.Conn, ethereum *Ethereum, inbound bool) *Peer {
 		caps:            ethereum.ServerCaps(),
 		version:         ethereum.ClientIdentity().String(),
 		protocolCaps:    ethutil.NewValue(nil),
+		td:              big.NewInt(0),
 	}
 }
 
@@ -192,6 +193,7 @@ func NewOutboundPeer(addr string, ethereum *Ethereum, caps Caps) *Peer {
 		caps:         caps,
 		version:      ethereum.ClientIdentity().String(),
 		protocolCaps: ethutil.NewValue(nil),
+		td:           big.NewInt(0),
 	}
 
 	// Set up the connection in another goroutine so we don't block the main thread
@@ -507,7 +509,7 @@ func (p *Peer) HandleInbound() {
 						hash := it.Value().Bytes()
 
 						p.lastReceivedHash = hash
-						p.lastHashReceived = time.Now()
+						p.LastHashReceived = time.Now()
 
 						if blockPool.HasCommonHash(hash) {
 							foundCommonHash = true
@@ -515,12 +517,10 @@ func (p *Peer) HandleInbound() {
 							break
 						}
 
-						blockPool.AddHash(hash)
+						blockPool.AddHash(hash, p)
 					}
 
-					if foundCommonHash || msg.Data.Len() == 0 {
-						p.FetchBlocks()
-					} else {
+					if !foundCommonHash && msg.Data.Len() != 0 {
 						p.FetchHashes()
 					}
 
@@ -538,20 +538,6 @@ func (p *Peer) HandleInbound() {
 
 						p.lastBlockReceived = time.Now()
 					}
-
-					var err error
-					blockPool.CheckLinkAndProcess(func(block *ethchain.Block) {
-						err = p.ethereum.StateManager().Process(block, false)
-					})
-
-					if err != nil {
-						peerlogger.Infoln(err)
-					} /*else {
-						// Don't trigger if there's just one block.
-						if blockPool.Len() != 0 && msg.Data.Len() > 1 {
-							p.FetchBlocks()
-						}
-					}*/
 				}
 			}
 		}
@@ -560,10 +546,7 @@ func (p *Peer) HandleInbound() {
 	p.Stop()
 }
 
-func (self *Peer) FetchBlocks() {
-	blockPool := self.ethereum.blockPool
-
-	hashes := blockPool.Take(100, self)
+func (self *Peer) FetchBlocks(hashes [][]byte) {
 	if len(hashes) > 0 {
 		self.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlocksTy, ethutil.ByteSliceToInterface(hashes)))
 	}
@@ -572,7 +555,7 @@ func (self *Peer) FetchBlocks() {
 func (self *Peer) FetchHashes() {
 	blockPool := self.ethereum.blockPool
 
-	if self.td.Cmp(blockPool.td) >= 0 {
+	if self.td.Cmp(self.ethereum.HighestTDPeer()) >= 0 {
 		blockPool.td = self.td
 
 		if !blockPool.HasLatestHash() {
@@ -581,6 +564,10 @@ func (self *Peer) FetchHashes() {
 	}
 }
 
+func (self *Peer) FetchingHashes() bool {
+	return time.Since(self.LastHashReceived) < 5*time.Second
+}
+
 // General update method
 func (self *Peer) update() {
 	serviceTimer := time.NewTicker(100 * time.Millisecond)
@@ -592,19 +579,12 @@ out:
 			if self.IsCap("eth") {
 				var (
 					sinceBlock = time.Since(self.lastBlockReceived)
-					sinceHash  = time.Since(self.lastHashReceived)
+					sinceHash  = time.Since(self.LastHashReceived)
 				)
 
 				if sinceBlock > 5*time.Second && sinceHash > 5*time.Second {
 					self.catchingUp = false
 				}
-
-				if sinceHash > 10*time.Second && self.ethereum.blockPool.Len() != 0 {
-					// XXX While this is completely and utterly incorrect, in order to do anything on the test net is to do it this way
-					// Assume that when fetching hashes timeouts, we are done.
-					//self.FetchHashes()
-					self.FetchBlocks()
-				}
 			}
 		case <-self.quit:
 			break out
@@ -728,7 +708,7 @@ func (self *Peer) handleStatus(msg *ethwire.Msg) {
 	// Compare the total TD with the blockchain TD. If remote is higher
 	// fetch hashes from highest TD node.
 	if self.td.Cmp(self.ethereum.BlockChain().TD) > 0 {
-		self.ethereum.blockPool.AddHash(self.lastReceivedHash)
+		self.ethereum.blockPool.AddHash(self.lastReceivedHash, self)
 		self.FetchHashes()
 	}