Merge branch 'release/0.6.6'

This commit is contained in:
obscuren 2014-09-24 20:40:54 +02:00
commit bb4495722f
9 changed files with 207 additions and 94 deletions

View File

@ -1,17 +1,27 @@
package eth package eth
import ( import (
"bytes"
"container/list"
"math" "math"
"math/big" "math/big"
"sync" "sync"
"time"
"github.com/ethereum/eth-go/ethchain" "github.com/ethereum/eth-go/ethchain"
"github.com/ethereum/eth-go/ethlog"
"github.com/ethereum/eth-go/ethutil" "github.com/ethereum/eth-go/ethutil"
"github.com/ethereum/eth-go/ethwire"
) )
var poollogger = ethlog.NewLogger("[BPOOL]")
type block struct { type block struct {
from *Peer
peer *Peer peer *Peer
block *ethchain.Block block *ethchain.Block
reqAt time.Time
requested int
} }
type BlockPool struct { type BlockPool struct {
@ -23,6 +33,9 @@ type BlockPool struct {
pool map[string]*block pool map[string]*block
td *big.Int td *big.Int
quit chan bool
ChainLength, BlocksProcessed int
} }
func NewBlockPool(eth *Ethereum) *BlockPool { func NewBlockPool(eth *Ethereum) *BlockPool {
@ -30,6 +43,7 @@ func NewBlockPool(eth *Ethereum) *BlockPool {
eth: eth, eth: eth,
pool: make(map[string]*block), pool: make(map[string]*block),
td: ethutil.Big0, 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 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 { 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...) 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()) { if self.pool[hash] == nil && !self.eth.BlockChain().HasBlock(b.Hash()) {
self.hashPool = append(self.hashPool, 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 { } else if self.pool[hash] != nil {
self.pool[hash].block = b 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 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 { for _, item := range self.pool {
if item.block != nil { if item.block != nil {
blocks = append(blocks, item.block) 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) ethchain.BlockBy(ethchain.Number).Sort(blocks)
for _, block := range blocks { for _, block := range blocks {
if self.eth.BlockChain().HasBlock(block.PrevHash) { if self.eth.BlockChain().HasBlock(block.PrevHash) {
procAmount++
f(block) f(block)
hash := block.Hash() hash := block.Hash()
@ -84,23 +128,108 @@ func (self *BlockPool) CheckLinkAndProcess(f func(block *ethchain.Block)) {
} }
} }
}
func (self *BlockPool) Take(amount int, peer *Peer) (hashes [][]byte) {
self.mut.Lock()
defer self.mut.Unlock()
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++
}
}
return return
} }
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))))
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)
}
}
}
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})
}
}
}
}

View File

@ -250,15 +250,13 @@ func (sm *StateManager) Process(block *Block, dontReact bool) (err error) {
fk := append([]byte("bloom"), block.Hash()...) fk := append([]byte("bloom"), block.Hash()...)
sm.Ethereum.Db().Put(fk, filter.Bin()) 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 { if dontReact == false {
sm.Ethereum.Reactor().Post("newBlock", block) sm.Ethereum.Reactor().Post("newBlock", block)
state.Manifest().Reset() state.Manifest().Reset()
} }
sm.Ethereum.Broadcast(ethwire.MsgBlockTy, []interface{}{block.Value().Val})
sm.Ethereum.TxPool().RemoveInvalid(state) sm.Ethereum.TxPool().RemoveInvalid(state)
} else { } else {
statelogger.Errorln("total diff failed") statelogger.Errorln("total diff failed")

View File

@ -158,6 +158,9 @@ func (s *Ethereum) StateManager() *ethchain.StateManager {
func (s *Ethereum) TxPool() *ethchain.TxPool { func (s *Ethereum) TxPool() *ethchain.TxPool {
return s.txPool return s.txPool
} }
func (s *Ethereum) BlockPool() *BlockPool {
return s.blockPool
}
func (self *Ethereum) Db() ethutil.Database { func (self *Ethereum) Db() ethutil.Database {
return self.db return self.db
} }
@ -383,7 +386,6 @@ func (s *Ethereum) ReapDeadPeerHandler() {
// Start the ethereum // Start the ethereum
func (s *Ethereum) Start(seed bool) { func (s *Ethereum) Start(seed bool) {
s.reactor.Start() s.reactor.Start()
s.blockPool.Start()
// Bind to addr and port // Bind to addr and port
ln, err := net.Listen("tcp", ":"+s.Port) ln, err := net.Listen("tcp", ":"+s.Port)
if err != nil { if err != nil {
@ -504,6 +506,7 @@ func (s *Ethereum) Stop() {
s.stateManager.Stop() s.stateManager.Stop()
s.reactor.Flush() s.reactor.Flush()
s.reactor.Stop() s.reactor.Stop()
s.blockPool.Stop()
ethlogger.Infoln("Server stopped") ethlogger.Infoln("Server stopped")
close(s.shutdownChan) close(s.shutdownChan)

View File

@ -3,7 +3,6 @@ package ethminer
import ( import (
"bytes" "bytes"
"sort" "sort"
"time"
"github.com/ethereum/eth-go/ethchain" "github.com/ethereum/eth-go/ethchain"
"github.com/ethereum/eth-go/ethlog" "github.com/ethereum/eth-go/ethlog"
@ -136,12 +135,6 @@ func (miner *Miner) listener() {
} }
} }
default: default:
// This hack is only temporarily
if len(miner.txs) == 0 {
time.Sleep(2 * time.Second)
continue
}
miner.mineNewBlock() miner.mineNewBlock()
} }
} }

View File

@ -36,13 +36,6 @@ func NewJSBlock(block *ethchain.Block) *JSBlock {
ptxs = append(ptxs, *NewJSTx(tx)) 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) 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)} 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)}

View File

@ -196,8 +196,8 @@ func (t *Trie) Update(key, value string) {
} }
func (t *Trie) Get(key string) string { func (t *Trie) Get(key string) string {
t.mut.RLock() t.mut.Lock()
defer t.mut.RUnlock() defer t.mut.Unlock()
k := CompactHexDecode(key) k := CompactHexDecode(key)
c := ethutil.NewValue(t.getState(t.Root, k)) c := ethutil.NewValue(t.getState(t.Root, k))

View File

@ -2,7 +2,6 @@ package ethutil
import ( import (
"encoding/json" "encoding/json"
"fmt"
"reflect" "reflect"
) )
@ -10,6 +9,7 @@ import (
// 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 {
val interface{}
list reflect.Value list reflect.Value
Length int Length int
} }
@ -21,7 +21,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{list, list.Len()} return &List{t, list, list.Len()}
} }
func EmptyList() *List { func EmptyList() *List {
@ -30,17 +30,24 @@ 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() == 3 {
fmt.Println("get", i, self.list.Index(i).Interface())
}
if self.list.Len() > i { if self.list.Len() > i {
return self.list.Index(i).Interface() i := self.list.Index(i).Interface()
return i
} }
return nil 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 // 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{}) {

View File

@ -961,7 +961,6 @@ func (self *Message) Addr() []byte {
} }
func (self *Message) Exec(codeAddr []byte, caller ClosureRef) (ret []byte, err error) { 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 queue := self.vm.queue
self.vm.queue = list.New() self.vm.queue = list.New()

73
peer.go
View File

@ -131,6 +131,7 @@ type Peer struct {
// Last received pong message // Last received pong message
lastPong int64 lastPong int64
lastBlockReceived time.Time lastBlockReceived time.Time
LastHashReceived time.Time
host []byte host []byte
port uint16 port uint16
@ -176,6 +177,7 @@ func NewPeer(conn net.Conn, ethereum *Ethereum, inbound bool) *Peer {
caps: ethereum.ServerCaps(), caps: ethereum.ServerCaps(),
version: ethereum.ClientIdentity().String(), version: ethereum.ClientIdentity().String(),
protocolCaps: ethutil.NewValue(nil), protocolCaps: ethutil.NewValue(nil),
td: big.NewInt(0),
} }
} }
@ -191,6 +193,7 @@ func NewOutboundPeer(addr string, ethereum *Ethereum, caps Caps) *Peer {
caps: caps, caps: caps,
version: ethereum.ClientIdentity().String(), version: ethereum.ClientIdentity().String(),
protocolCaps: ethutil.NewValue(nil), protocolCaps: ethutil.NewValue(nil),
td: big.NewInt(0),
} }
// Set up the connection in another goroutine so we don't block the main thread // 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() { for it.Next() {
hash := it.Value().Bytes() hash := it.Value().Bytes()
p.lastReceivedHash = hash
p.LastHashReceived = time.Now()
if blockPool.HasCommonHash(hash) { if blockPool.HasCommonHash(hash) {
foundCommonHash = true foundCommonHash = true
break break
} }
blockPool.AddHash(hash) blockPool.AddHash(hash, p)
p.lastReceivedHash = hash
p.lastBlockReceived = time.Now()
} }
if foundCommonHash || msg.Data.Len() == 0 { if !foundCommonHash && msg.Data.Len() != 0 {
p.FetchBlocks()
} else {
p.FetchHashes() p.FetchHashes()
} }
@ -538,20 +538,6 @@ func (p *Peer) HandleInbound() {
p.lastBlockReceived = time.Now() 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() p.Stop()
} }
func (self *Peer) FetchBlocks() { func (self *Peer) FetchBlocks(hashes [][]byte) {
blockPool := self.ethereum.blockPool
hashes := blockPool.Take(100, self)
if len(hashes) > 0 { if len(hashes) > 0 {
self.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlocksTy, ethutil.ByteSliceToInterface(hashes))) self.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlocksTy, ethutil.ByteSliceToInterface(hashes)))
} }
@ -572,7 +555,7 @@ func (self *Peer) FetchBlocks() {
func (self *Peer) FetchHashes() { func (self *Peer) FetchHashes() {
blockPool := self.ethereum.blockPool blockPool := self.ethereum.blockPool
if self.td.Cmp(blockPool.td) >= 0 { if self.td.Cmp(self.ethereum.HighestTDPeer()) >= 0 {
blockPool.td = self.td blockPool.td = self.td
if !blockPool.HasLatestHash() { if !blockPool.HasLatestHash() {
@ -581,20 +564,28 @@ func (self *Peer) FetchHashes() {
} }
} }
func (self *Peer) FetchingHashes() bool {
return time.Since(self.LastHashReceived) < 5*time.Second
}
// General update method // General update method
func (self *Peer) update() { func (self *Peer) update() {
serviceTimer := time.NewTicker(5 * time.Second) serviceTimer := time.NewTicker(100 * time.Millisecond)
out: out:
for { for {
select { select {
case <-serviceTimer.C: case <-serviceTimer.C:
since := time.Since(self.lastBlockReceived) if self.IsCap("eth") {
if since > 10*time.Second && self.ethereum.blockPool.Len() != 0 && self.IsCap("eth") { var (
self.FetchHashes() sinceBlock = time.Since(self.lastBlockReceived)
} else if since > 5*time.Second { sinceHash = time.Since(self.LastHashReceived)
)
if sinceBlock > 5*time.Second && sinceHash > 5*time.Second {
self.catchingUp = false self.catchingUp = false
} }
}
case <-self.quit: case <-self.quit:
break out 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 // Compare the total TD with the blockchain TD. If remote is higher
// fetch hashes from highest TD node. // fetch hashes from highest TD node.
if self.td.Cmp(self.ethereum.BlockChain().TD) > 0 { if self.td.Cmp(self.ethereum.BlockChain().TD) > 0 {
self.ethereum.blockPool.AddHash(self.lastReceivedHash) self.ethereum.blockPool.AddHash(self.lastReceivedHash, self)
self.FetchHashes() self.FetchHashes()
} }
@ -761,6 +752,14 @@ func (p *Peer) handleHandshake(msg *ethwire.Msg) {
return return
} }
// Self connect detection
pubkey := p.ethereum.KeyManager().PublicKey()
if bytes.Compare(pubkey[1:], pub) == 0 {
p.Stop()
return
}
usedPub := 0 usedPub := 0
// This peer is already added to the peerlist so we expect to find a double pubkey at least once // 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) { 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 this is an inbound connection send an ack back
if p.inbound { if p.inbound {
p.port = uint16(port) 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.SetVersion(clientId)
p.versionKnown = true p.versionKnown = true