Block pool is thread safe

pull/150/head
obscuren 10 years ago
parent 44d50bc8d2
commit ea0357bf02
  1. 101
      block_pool.go
  2. 10
      ethchain/state_manager.go
  3. 11
      peer.go

@ -1,7 +1,6 @@
package eth package eth
import ( import (
"bytes"
"container/list" "container/list"
"math" "math"
"math/big" "math/big"
@ -35,6 +34,9 @@ type BlockPool struct {
td *big.Int td *big.Int
quit chan bool quit chan bool
fetchingHashes bool
downloadStartedAt time.Time
ChainLength, BlocksProcessed int ChainLength, BlocksProcessed int
} }
@ -52,6 +54,9 @@ func (self *BlockPool) Len() int {
} }
func (self *BlockPool) HasLatestHash() bool { func (self *BlockPool) HasLatestHash() bool {
self.mut.Lock()
defer self.mut.Unlock()
return self.pool[string(self.eth.BlockChain().CurrentBlock.Hash())] != nil return self.pool[string(self.eth.BlockChain().CurrentBlock.Hash())] != nil
} }
@ -59,7 +64,20 @@ func (self *BlockPool) HasCommonHash(hash []byte) bool {
return self.eth.BlockChain().GetBlock(hash) != nil return self.eth.BlockChain().GetBlock(hash) != nil
} }
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) AddHash(hash []byte, peer *Peer) { func (self *BlockPool) AddHash(hash []byte, peer *Peer) {
self.mut.Lock()
defer self.mut.Unlock()
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}
@ -67,7 +85,10 @@ func (self *BlockPool) AddHash(hash []byte, peer *Peer) {
} }
} }
func (self *BlockPool) SetBlock(b *ethchain.Block, peer *Peer) { func (self *BlockPool) Add(b *ethchain.Block, peer *Peer) {
self.mut.Lock()
defer self.mut.Unlock()
hash := string(b.Hash()) hash := string(b.Hash())
if self.pool[hash] == nil && !self.eth.BlockChain().HasBlock(b.Hash()) { if self.pool[hash] == nil && !self.eth.BlockChain().HasBlock(b.Hash()) {
@ -76,7 +97,7 @@ func (self *BlockPool) SetBlock(b *ethchain.Block, peer *Peer) {
self.hashPool = append(self.hashPool, b.Hash()) self.hashPool = append(self.hashPool, b.Hash())
self.pool[hash] = &block{peer, peer, b, time.Now(), 0} self.pool[hash] = &block{peer, peer, b, time.Now(), 0}
if !self.eth.BlockChain().HasBlock(b.PrevHash) { if !self.eth.BlockChain().HasBlock(b.PrevHash) && !self.fetchingHashes {
poollogger.Infof("Unknown block, requesting parent (%x...)\n", b.PrevHash[0:4]) poollogger.Infof("Unknown block, requesting parent (%x...)\n", b.PrevHash[0:4])
peer.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlockHashesTy, []interface{}{b.PrevHash, uint32(256)})) peer.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlockHashesTy, []interface{}{b.PrevHash, uint32(256)}))
} }
@ -87,36 +108,12 @@ func (self *BlockPool) SetBlock(b *ethchain.Block, peer *Peer) {
self.BlocksProcessed++ self.BlocksProcessed++
} }
func (self *BlockPool) getParent(block *ethchain.Block) *ethchain.Block { func (self *BlockPool) Remove(hash []byte) {
for _, item := range self.pool { self.mut.Lock()
if item.block != nil { defer self.mut.Unlock()
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 self.hashPool = ethutil.DeleteFromByteSlice(self.hashPool, hash)
delete(self.pool, string(hash))
} }
func (self *BlockPool) ProcessCanonical(f func(block *ethchain.Block)) (procAmount int) { func (self *BlockPool) ProcessCanonical(f func(block *ethchain.Block)) (procAmount int) {
@ -129,9 +126,7 @@ func (self *BlockPool) ProcessCanonical(f func(block *ethchain.Block)) (procAmou
f(block) f(block)
hash := block.Hash() self.Remove(block.Hash())
self.hashPool = ethutil.DeleteFromByteSlice(self.hashPool, hash)
delete(self.pool, string(hash))
} }
} }
@ -140,9 +135,12 @@ func (self *BlockPool) ProcessCanonical(f func(block *ethchain.Block)) (procAmou
} }
func (self *BlockPool) DistributeHashes() { func (self *BlockPool) DistributeHashes() {
self.mut.Lock()
defer self.mut.Unlock()
var ( var (
peerLen = self.eth.peers.Len() peerLen = self.eth.peers.Len()
amount = 200 * peerLen amount = 256 * peerLen
dist = make(map[*Peer][][]byte) dist = make(map[*Peer][][]byte)
) )
@ -156,7 +154,7 @@ func (self *BlockPool) DistributeHashes() {
lastFetchFailed := time.Since(item.reqAt) > 5*time.Second lastFetchFailed := time.Since(item.reqAt) > 5*time.Second
// Handle failed requests // Handle failed requests
if lastFetchFailed && item.requested > 0 && item.peer != nil { if lastFetchFailed && item.requested > 5 && item.peer != nil {
if item.requested < 100 { if item.requested < 100 {
// Select peer the hash was retrieved off // Select peer the hash was retrieved off
peer = item.from peer = item.from
@ -187,19 +185,23 @@ func (self *BlockPool) DistributeHashes() {
for peer, hashes := range dist { for peer, hashes := range dist {
peer.FetchBlocks(hashes) peer.FetchBlocks(hashes)
} }
if len(dist) > 0 {
self.downloadStartedAt = time.Now()
}
} }
func (self *BlockPool) Start() { func (self *BlockPool) Start() {
go self.update() go self.downloadThread()
go self.chainThread()
} }
func (self *BlockPool) Stop() { func (self *BlockPool) Stop() {
close(self.quit) close(self.quit)
} }
func (self *BlockPool) update() { func (self *BlockPool) downloadThread() {
serviceTimer := time.NewTicker(100 * time.Millisecond) serviceTimer := time.NewTicker(100 * time.Millisecond)
procTimer := time.NewTicker(500 * time.Millisecond)
out: out:
for { for {
select { select {
@ -208,20 +210,31 @@ out:
case <-serviceTimer.C: case <-serviceTimer.C:
// Check if we're catching up. If not distribute the hashes to // Check if we're catching up. If not distribute the hashes to
// the peers and download the blockchain // the peers and download the blockchain
done := true self.fetchingHashes = false
eachPeer(self.eth.peers, func(p *Peer, v *list.Element) { eachPeer(self.eth.peers, func(p *Peer, v *list.Element) {
if p.statusKnown && p.FetchingHashes() { if p.statusKnown && p.FetchingHashes() {
done = false self.fetchingHashes = true
} }
}) })
if done && len(self.hashPool) > 0 { if !self.fetchingHashes && len(self.hashPool) > 0 {
self.DistributeHashes() self.DistributeHashes()
} }
if self.ChainLength < len(self.hashPool) { if self.ChainLength < len(self.hashPool) {
self.ChainLength = len(self.hashPool) self.ChainLength = len(self.hashPool)
} }
}
}
}
func (self *BlockPool) chainThread() {
procTimer := time.NewTicker(500 * time.Millisecond)
out:
for {
select {
case <-self.quit:
break out
case <-procTimer.C: case <-procTimer.C:
// XXX We can optimize this lifting this on to a new goroutine. // 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 // We'd need to make sure that the pools are properly protected by a mutex
@ -230,6 +243,8 @@ out:
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)
poollogger.Debugf("Block #%v failed (%x...)\n", block.Number, block.Hash()[0:4])
poollogger.Debugln(block)
} }
}) })

@ -143,9 +143,6 @@ done:
} }
} }
// Notify all subscribers
self.Ethereum.Reactor().Post("newTx:post", tx)
// Update the state with pending changes // Update the state with pending changes
state.Update() state.Update()
@ -160,10 +157,15 @@ done:
os.Exit(1) os.Exit(1)
} }
return nil, nil, nil, fmt.Errorf("err diff #%d (r) %v ~ %x <=> (c) %v ~ %x (%x)\n", i+1, original.CumulativeGasUsed, original.PostState[0:4], receipt.CumulativeGasUsed, receipt.PostState[0:4], receipt.Tx.Hash()) err := fmt.Errorf("#%d receipt failed (r) %v ~ %x <=> (c) %v ~ %x (%x...)", i+1, original.CumulativeGasUsed, original.PostState[0:4], receipt.CumulativeGasUsed, receipt.PostState[0:4], receipt.Tx.Hash()[0:4])
return nil, nil, nil, err
} }
} }
// Notify all subscribers
self.Ethereum.Reactor().Post("newTx:post", tx)
receipts = append(receipts, receipt) receipts = append(receipts, receipt)
handled = append(handled, tx) handled = append(handled, tx)

@ -503,16 +503,15 @@ func (p *Peer) HandleInbound() {
it := msg.Data.NewIterator() it := msg.Data.NewIterator()
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
} }
p.lastReceivedHash = hash
p.LastHashReceived = time.Now()
blockPool.AddHash(hash, p) blockPool.AddHash(hash, p)
} }
@ -530,7 +529,7 @@ func (p *Peer) HandleInbound() {
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]) //fmt.Printf("%v %x - %x\n", block.Number, block.Hash()[0:4], block.PrevHash[0:4])
blockPool.SetBlock(block, p) blockPool.Add(block, p)
p.lastBlockReceived = time.Now() p.lastBlockReceived = time.Now()
} }
@ -561,7 +560,7 @@ func (self *Peer) FetchHashes() {
} }
func (self *Peer) FetchingHashes() bool { func (self *Peer) FetchingHashes() bool {
return time.Since(self.LastHashReceived) < 5*time.Second return time.Since(self.LastHashReceived) < 200*time.Millisecond
} }
// General update method // General update method

Loading…
Cancel
Save