aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorobscuren <geffobscura@gmail.com>2014-09-25 02:40:54 +0800
committerobscuren <geffobscura@gmail.com>2014-09-25 02:40:54 +0800
commitbb4495722f869f456f1f12234bb881ef71a7960f (patch)
treeae9b4ddf48556081ff4d3d3328dbc4227f4aa256
parent57dc435f9b928f5de2a49736a2c71a7bf611289a (diff)
parent1118aaf840a6f6b4dd6b137f39ab895a0cbd5a56 (diff)
downloadgo-tangerine-bb4495722f869f456f1f12234bb881ef71a7960f.tar.gz
go-tangerine-bb4495722f869f456f1f12234bb881ef71a7960f.tar.zst
go-tangerine-bb4495722f869f456f1f12234bb881ef71a7960f.zip
Merge branch 'release/0.6.6'
-rw-r--r--block_pool.go167
-rw-r--r--ethchain/state_manager.go4
-rw-r--r--ethereum.go5
-rw-r--r--ethminer/miner.go7
-rw-r--r--ethpipe/js_types.go7
-rw-r--r--ethtrie/trie.go4
-rw-r--r--ethutil/list.go19
-rw-r--r--ethvm/vm.go1
-rw-r--r--peer.go77
9 files changed, 202 insertions, 89 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 09665336c..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
}
@@ -383,7 +386,6 @@ func (s *Ethereum) ReapDeadPeerHandler() {
// Start the ethereum
func (s *Ethereum) Start(seed bool) {
s.reactor.Start()
- s.blockPool.Start()
// Bind to addr and port
ln, err := net.Listen("tcp", ":"+s.Port)
if err != nil {
@@ -504,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/ethpipe/js_types.go b/ethpipe/js_types.go
index 8d2805f3d..ccd585cf0 100644
--- a/ethpipe/js_types.go
+++ b/ethpipe/js_types.go
@@ -36,13 +36,6 @@ func NewJSBlock(block *ethchain.Block) *JSBlock {
ptxs = append(ptxs, *NewJSTx(tx))
}
- /*
- txJson, err := json.Marshal(ptxs)
- if err != nil {
- return nil
- }
- return &JSBlock{ref: block, Size: block.Size().String(), Number: int(block.Number.Uint64()), GasUsed: block.GasUsed.String(), GasLimit: block.GasLimit.String(), Hash: ethutil.Bytes2Hex(block.Hash()), Transactions: string(txJson), Time: block.Time, Coinbase: ethutil.Bytes2Hex(block.Coinbase)}
- */
list := ethutil.NewList(ptxs)
return &JSBlock{ref: block, Size: block.Size().String(), Number: int(block.Number.Uint64()), GasUsed: block.GasUsed.String(), GasLimit: block.GasLimit.String(), Hash: ethutil.Bytes2Hex(block.Hash()), Transactions: list, Time: block.Time, Coinbase: ethutil.Bytes2Hex(block.Coinbase)}
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/ethutil/list.go b/ethutil/list.go
index a5147573a..4fb36224f 100644
--- a/ethutil/list.go
+++ b/ethutil/list.go
@@ -2,7 +2,6 @@ package ethutil
import (
"encoding/json"
- "fmt"
"reflect"
)
@@ -10,6 +9,7 @@ import (
// for containing any slice type to use in an environment which
// does not support slice types (e.g., JavaScript, QML)
type List struct {
+ val interface{}
list reflect.Value
Length int
}
@@ -21,7 +21,7 @@ func NewList(t interface{}) *List {
panic("list container initialized with a non-slice type")
}
- return &List{list, list.Len()}
+ return &List{t, list, list.Len()}
}
func EmptyList() *List {
@@ -30,17 +30,24 @@ 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() == 3 {
- fmt.Println("get", i, self.list.Index(i).Interface())
- }
if self.list.Len() > i {
- return self.list.Index(i).Interface()
+ i := self.list.Index(i).Interface()
+
+ return i
}
return nil
}
+func (self *List) GetAsJson(i int) interface{} {
+ e := self.Get(i)
+
+ r, _ := json.Marshal(e)
+
+ return string(r)
+}
+
// Appends value at the end of the slice. Panics when incompatible value
// is given.
func (self *List) Append(v interface{}) {
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 529a35492..06df5fab6 100644
--- a/peer.go
+++ b/peer.go
@@ -131,6 +131,7 @@ type Peer struct {
// Last received pong message
lastPong int64
lastBlockReceived time.Time
+ LastHashReceived time.Time
host []byte
port uint16
@@ -176,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),
}
}
@@ -191,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
@@ -505,22 +508,19 @@ func (p *Peer) HandleInbound() {
for it.Next() {
hash := it.Value().Bytes()
+ p.lastReceivedHash = hash
+ p.LastHashReceived = time.Now()
+
if blockPool.HasCommonHash(hash) {
foundCommonHash = true
break
}
- blockPool.AddHash(hash)
-
- p.lastReceivedHash = hash
-
- p.lastBlockReceived = time.Now()
+ 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,19 +564,27 @@ 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(5 * time.Second)
+ serviceTimer := time.NewTicker(100 * time.Millisecond)
out:
for {
select {
case <-serviceTimer.C:
- since := time.Since(self.lastBlockReceived)
- if since > 10*time.Second && self.ethereum.blockPool.Len() != 0 && self.IsCap("eth") {
- self.FetchHashes()
- } else if since > 5*time.Second {
- self.catchingUp = false
+ if self.IsCap("eth") {
+ var (
+ sinceBlock = time.Since(self.lastBlockReceived)
+ sinceHash = time.Since(self.LastHashReceived)
+ )
+
+ if sinceBlock > 5*time.Second && sinceHash > 5*time.Second {
+ self.catchingUp = false
+ }
}
case <-self.quit:
break out
@@ -717,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()
}
@@ -761,6 +752,14 @@ func (p *Peer) handleHandshake(msg *ethwire.Msg) {
return
}
+ // Self connect detection
+ pubkey := p.ethereum.KeyManager().PublicKey()
+ if bytes.Compare(pubkey[1:], pub) == 0 {
+ p.Stop()
+
+ return
+ }
+
usedPub := 0
// This peer is already added to the peerlist so we expect to find a double pubkey at least once
eachPeer(p.ethereum.Peers(), func(peer *Peer, e *list.Element) {
@@ -779,16 +778,8 @@ func (p *Peer) handleHandshake(msg *ethwire.Msg) {
// If this is an inbound connection send an ack back
if p.inbound {
p.port = uint16(port)
-
- // Self connect detection
- pubkey := p.ethereum.KeyManager().PublicKey()
- if bytes.Compare(pubkey, p.pubkey) == 0 {
- p.Stop()
-
- return
- }
-
}
+
p.SetVersion(clientId)
p.versionKnown = true