mirror
/
go-ethereum
mirror of https://github.com/ethereum/go-ethereum
2
0
Fork 1
Code Issues Releases Wiki Activity

Merge branch 'ethersphere-develop' into develop

Browse Source
pull/253/head
obscuren 10 years ago
parent 6eb455032c a35be3a12a
commit b136b387a7
8 changed files with 500 additions and 191 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 4
      core/chain_manager.go
  2. 435
      eth/block_pool.go
  3. 188
      eth/block_pool_test.go
  4. 2
      eth/error.go
  5. 10
      eth/protocol.go
  6. 6
      eth/test/tests/02.sh
  7. 4
      eth/test/tests/03.sh
  8. 6
      eth/test/tests/common.sh

4
core/chain_manager.go
Unescape View File

@ -238,13 +238,11 @@ func (self *ChainManager) GetBlockHashesFromHash(hash []byte, max uint64) (chain
// XXX Could be optimised by using a different database which only holds hashes (i.e., linked list) // XXX Could be optimised by using a different database which only holds hashes (i.e., linked list)
for i := uint64(0); i < max; i++ { for i := uint64(0); i < max; i++ {
block = self.GetBlock(block.Header().ParentHash)
chain = append(chain, block.Hash()) chain = append(chain, block.Hash())
if block.Header().Number.Cmp(ethutil.Big0) <= 0 { if block.Header().Number.Cmp(ethutil.Big0) <= 0 {
break break
} }
block = self.GetBlock(block.Header().ParentHash)
} }
return return

435
eth/block_pool.go
Unescape View File

@ -1,6 +1,7 @@
package eth package eth
import ( import (
"bytes"
"fmt" "fmt"
"math" "math"
"math/big" "math/big"
@ -24,8 +25,8 @@ const (
blocksRequestRepetition = 1 blocksRequestRepetition = 1
blockHashesRequestInterval = 500 // ms blockHashesRequestInterval = 500 // ms
blocksRequestMaxIdleRounds = 100 blocksRequestMaxIdleRounds = 100
cacheTimeout = 3 // minutes blockHashesTimeout = 60 // seconds
blockTimeout = 5 // minutes blocksTimeout = 120 // seconds
) )
type poolNode struct { type poolNode struct {
@ -71,7 +72,12 @@ type peerInfo struct {
lock sync.RWMutex lock sync.RWMutex
td *big.Int td *big.Int
currentBlock []byte currentBlockHash []byte
currentBlock *types.Block
currentBlockC chan *types.Block
parentHash []byte
headSection *section
headSectionC chan *section
id string id string
requestBlockHashes func([]byte) error requestBlockHashes func([]byte) error
@ -203,30 +209,39 @@ func (self *BlockPool) Wait(t time.Duration) {
// AddPeer is called by the eth protocol instance running on the peer after // AddPeer is called by the eth protocol instance running on the peer after
// the status message has been received with total difficulty and current block hash // the status message has been received with total difficulty and current block hash
// AddPeer can only be used once, RemovePeer needs to be called when the peer disconnects // AddPeer can only be used once, RemovePeer needs to be called when the peer disconnects
func (self *BlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) bool { func (self *BlockPool) AddPeer(td *big.Int, currentBlockHash []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) (best bool) {
self.peersLock.Lock() self.peersLock.Lock()
defer self.peersLock.Unlock() defer self.peersLock.Unlock()
peer, ok := self.peers[peerId] peer, ok := self.peers[peerId]
if ok { if ok {
poolLogger.Debugf("Update peer %v with td %v and current block %x", peerId, td, currentBlock[:4]) if bytes.Compare(peer.currentBlockHash, currentBlockHash) != 0 {
poolLogger.Debugf("Update peer %v with td %v and current block %s", peerId, td, name(currentBlockHash))
peer.lock.Lock()
peer.td = td peer.td = td
peer.currentBlock = currentBlock peer.currentBlockHash = currentBlockHash
peer.currentBlock = nil
peer.parentHash = nil
peer.headSection = nil
peer.lock.Unlock()
}
} else { } else {
peer = &peerInfo{ peer = &peerInfo{
td: td, td: td,
currentBlock: currentBlock, currentBlockHash: currentBlockHash,
id: peerId, //peer.Identity().Pubkey() id: peerId, //peer.Identity().Pubkey()
requestBlockHashes: requestBlockHashes, requestBlockHashes: requestBlockHashes,
requestBlocks: requestBlocks, requestBlocks: requestBlocks,
peerError: peerError, peerError: peerError,
sections: make(map[string]*section), sections: make(map[string]*section),
currentBlockC: make(chan *types.Block),
headSectionC: make(chan *section),
} }
self.peers[peerId] = peer self.peers[peerId] = peer
poolLogger.Debugf("add new peer %v with td %v and current block %x", peerId, td, currentBlock[:4]) poolLogger.Debugf("add new peer %v with td %v and current block %x", peerId, td, currentBlockHash[:4])
} }
// check peer current head // check peer current head
if self.hasBlock(currentBlock) { if self.hasBlock(currentBlockHash) {
// peer not ahead // peer not ahead
return false return false
} }
@ -234,11 +249,10 @@ func (self *BlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string,
if self.peer == peer { if self.peer == peer {
// new block update // new block update
// peer is already active best peer, request hashes // peer is already active best peer, request hashes
poolLogger.Debugf("[%s] already the best peer. request hashes from %s", peerId, name(currentBlock)) poolLogger.Debugf("[%s] already the best peer. Request new head section info from %s", peerId, name(currentBlockHash))
peer.requestBlockHashes(currentBlock) peer.headSectionC <- nil
return true best = true
} } else {
currentTD := ethutil.Big0 currentTD := ethutil.Big0
if self.peer != nil { if self.peer != nil {
currentTD = self.peer.td currentTD = self.peer.td
@ -247,9 +261,123 @@ func (self *BlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string,
poolLogger.Debugf("peer %v promoted best peer", peerId) poolLogger.Debugf("peer %v promoted best peer", peerId)
self.switchPeer(self.peer, peer) self.switchPeer(self.peer, peer)
self.peer = peer self.peer = peer
return true best = true
} }
return false }
return
}
func (self *BlockPool) requestHeadSection(peer *peerInfo) {
self.wg.Add(1)
self.procWg.Add(1)
poolLogger.Debugf("[%s] head section at [%s] requesting info", peer.id, name(peer.currentBlockHash))
go func() {
var idle bool
peer.lock.RLock()
quitC := peer.quitC
currentBlockHash := peer.currentBlockHash
peer.lock.RUnlock()
blockHashesRequestTimer := time.NewTimer(0)
blocksRequestTimer := time.NewTimer(0)
suicide := time.NewTimer(blockHashesTimeout * time.Second)
blockHashesRequestTimer.Stop()
defer blockHashesRequestTimer.Stop()
defer blocksRequestTimer.Stop()
entry := self.get(currentBlockHash)
if entry != nil {
entry.node.lock.RLock()
currentBlock := entry.node.block
entry.node.lock.RUnlock()
if currentBlock != nil {
peer.lock.Lock()
peer.currentBlock = currentBlock
peer.parentHash = currentBlock.ParentHash()
poolLogger.Debugf("[%s] head block [%s] found", peer.id, name(currentBlockHash))
peer.lock.Unlock()
blockHashesRequestTimer.Reset(0)
blocksRequestTimer.Stop()
}
}
LOOP:
for {
select {
case <-self.quit:
break LOOP
case <-quitC:
poolLogger.Debugf("[%s] head section at [%s] incomplete - quit request loop", peer.id, name(currentBlockHash))
break LOOP
case headSection := <-peer.headSectionC:
peer.lock.Lock()
peer.headSection = headSection
if headSection == nil {
oldBlockHash := currentBlockHash
currentBlockHash = peer.currentBlockHash
poolLogger.Debugf("[%s] head section changed [%s] -> [%s]", peer.id, name(oldBlockHash), name(currentBlockHash))
if idle {
idle = false
suicide.Reset(blockHashesTimeout * time.Second)
self.procWg.Add(1)
}
blocksRequestTimer.Reset(blocksRequestInterval * time.Millisecond)
} else {
poolLogger.DebugDetailf("[%s] head section at [%s] created", peer.id, name(currentBlockHash))
if !idle {
idle = true
suicide.Stop()
self.procWg.Done()
}
}
peer.lock.Unlock()
blockHashesRequestTimer.Stop()
case <-blockHashesRequestTimer.C:
poolLogger.DebugDetailf("[%s] head section at [%s] not found, requesting block hashes", peer.id, name(currentBlockHash))
peer.requestBlockHashes(currentBlockHash)
blockHashesRequestTimer.Reset(blockHashesRequestInterval * time.Millisecond)
case currentBlock := <-peer.currentBlockC:
peer.lock.Lock()
peer.currentBlock = currentBlock
peer.parentHash = currentBlock.ParentHash()
poolLogger.DebugDetailf("[%s] head block [%s] found", peer.id, name(currentBlockHash))
peer.lock.Unlock()
if self.hasBlock(currentBlock.ParentHash()) {
if err := self.insertChain(types.Blocks([]*types.Block{currentBlock})); err != nil {
peer.peerError(ErrInvalidBlock, "%v", err)
}
if !idle {
idle = true
suicide.Stop()
self.procWg.Done()
}
} else {
blockHashesRequestTimer.Reset(0)
}
blocksRequestTimer.Stop()
case <-blocksRequestTimer.C:
peer.lock.RLock()
poolLogger.DebugDetailf("[%s] head block [%s] not found, requesting", peer.id, name(currentBlockHash))
peer.requestBlocks([][]byte{peer.currentBlockHash})
peer.lock.RUnlock()
blocksRequestTimer.Reset(blocksRequestInterval * time.Millisecond)
case <-suicide.C:
peer.peerError(ErrInsufficientChainInfo, "peer failed to provide block hashes or head block for block hash %x", currentBlockHash)
break LOOP
}
}
self.wg.Done()
if !idle {
self.procWg.Done()
}
}()
} }
// RemovePeer is called by the eth protocol when the peer disconnects // RemovePeer is called by the eth protocol when the peer disconnects
@ -274,13 +402,13 @@ func (self *BlockPool) RemovePeer(peerId string) {
newPeer = info newPeer = info
} }
} }
self.peer = newPeer
self.switchPeer(peer, newPeer)
if newPeer != nil { if newPeer != nil {
poolLogger.Debugf("peer %v with td %v promoted to best peer", newPeer.id, newPeer.td) poolLogger.Debugf("peer %v with td %v promoted to best peer", newPeer.id, newPeer.td)
} else { } else {
poolLogger.Warnln("no peers") poolLogger.Warnln("no peers")
} }
self.peer = newPeer
self.switchPeer(peer, newPeer)
} }
} }
@ -299,25 +427,56 @@ func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string)
return return
} }
// peer is still the best // peer is still the best
poolLogger.Debugf("adding hashes for best peer %s", peerId)
var size, n int var size, n int
var hash []byte var hash []byte
var ok bool var ok, headSection bool
var section, child, parent *section var sec, child, parent *section
var entry *poolEntry var entry *poolEntry
var nodes []*poolNode var nodes []*poolNode
bestPeer := peer
hash, ok = next()
peer.lock.Lock()
if bytes.Compare(peer.parentHash, hash) == 0 {
if self.hasBlock(peer.currentBlockHash) {
return
}
poolLogger.Debugf("adding hashes at chain head for best peer %s starting from [%s]", peerId, name(peer.currentBlockHash))
headSection = true
if entry := self.get(peer.currentBlockHash); entry == nil {
node := &poolNode{
hash: peer.currentBlockHash,
block: peer.currentBlock,
peer: peerId,
blockBy: peerId,
}
if size == 0 {
sec = newSection()
}
nodes = append(nodes, node)
size++
n++
} else {
child = entry.section
}
} else {
poolLogger.Debugf("adding hashes for best peer %s starting from [%s]", peerId, name(hash))
}
quitC := peer.quitC
peer.lock.Unlock()
LOOP: LOOP:
// iterate using next (rlp stream lazy decoder) feeding hashesC // iterate using next (rlp stream lazy decoder) feeding hashesC
for hash, ok = next(); ok; hash, ok = next() { for ; ok; hash, ok = next() {
n++ n++
select { select {
case <-self.quit: case <-self.quit:
return return
case <-peer.quitC: case <-quitC:
// if the peer is demoted, no more hashes taken // if the peer is demoted, no more hashes taken
peer = nil bestPeer = nil
break LOOP break LOOP
default: default:
} }
@ -325,8 +484,8 @@ LOOP:
// check if known block connecting the downloaded chain to our blockchain // check if known block connecting the downloaded chain to our blockchain
poolLogger.DebugDetailf("[%s] known block", name(hash)) poolLogger.DebugDetailf("[%s] known block", name(hash))
// mark child as absolute pool root with parent known to blockchain // mark child as absolute pool root with parent known to blockchain
if section != nil { if sec != nil {
self.connectToBlockChain(section) self.connectToBlockChain(sec)
} else { } else {
if child != nil { if child != nil {
self.connectToBlockChain(child) self.connectToBlockChain(child)
@ -340,6 +499,7 @@ LOOP:
// reached a known chain in the pool // reached a known chain in the pool
if entry.node == entry.section.bottom && n == 1 { if entry.node == entry.section.bottom && n == 1 {
// the first block hash received is an orphan in the pool, so rejoice and continue // the first block hash received is an orphan in the pool, so rejoice and continue
poolLogger.DebugDetailf("[%s] connecting child section", sectionName(entry.section))
child = entry.section child = entry.section
continue LOOP continue LOOP
} }
@ -353,7 +513,7 @@ LOOP:
peer: peerId, peer: peerId,
} }
if size == 0 { if size == 0 {
section = newSection() sec = newSection()
} }
nodes = append(nodes, node) nodes = append(nodes, node)
size++ size++
@ -379,10 +539,10 @@ LOOP:
} }
if size > 0 { if size > 0 {
self.processSection(section, nodes) self.processSection(sec, nodes)
poolLogger.DebugDetailf("[%s]->[%s](%v)->[%s] new chain section", sectionName(parent), sectionName(section), size, sectionName(child)) poolLogger.DebugDetailf("[%s]->[%s](%v)->[%s] new chain section", sectionName(parent), sectionName(sec), size, sectionName(child))
self.link(parent, section) self.link(parent, sec)
self.link(section, child) self.link(sec, child)
} else { } else {
poolLogger.DebugDetailf("[%s]->[%s] connecting known sections", sectionName(parent), sectionName(child)) poolLogger.DebugDetailf("[%s]->[%s] connecting known sections", sectionName(parent), sectionName(child))
self.link(parent, child) self.link(parent, child)
@ -390,15 +550,31 @@ LOOP:
self.chainLock.Unlock() self.chainLock.Unlock()
if parent != nil && peer != nil { if parent != nil && bestPeer != nil {
self.activateChain(parent, peer) self.activateChain(parent, peer)
poolLogger.Debugf("[%s] activate parent section [%s]", name(parent.top.hash), sectionName(parent)) poolLogger.Debugf("[%s] activate parent section [%s]", name(parent.top.hash), sectionName(parent))
} }
if section != nil { if sec != nil {
peer.addSection(section.top.hash, section) peer.addSection(sec.top.hash, sec)
section.controlC <- peer // request next section here once, only repeat if bottom block arrives,
poolLogger.Debugf("[%s] activate new section", sectionName(section)) // otherwise no way to check if it arrived
peer.requestBlockHashes(sec.bottom.hash)
sec.controlC <- bestPeer
poolLogger.Debugf("[%s] activate new section", sectionName(sec))
}
if headSection {
var headSec *section
switch {
case sec != nil:
headSec = sec
case child != nil:
headSec = child
default:
headSec = parent
}
peer.headSectionC <- headSec
} }
} }
@ -426,14 +602,21 @@ func sectionName(section *section) (name string) {
// only the first PoW-valid block for a hash is considered legit // only the first PoW-valid block for a hash is considered legit
func (self *BlockPool) AddBlock(block *types.Block, peerId string) { func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
hash := block.Hash() hash := block.Hash()
if self.hasBlock(hash) { self.peersLock.Lock()
poolLogger.DebugDetailf("block [%s] already known", name(hash)) peer := self.peer
return self.peersLock.Unlock()
}
entry := self.get(hash) entry := self.get(hash)
if bytes.Compare(hash, peer.currentBlockHash) == 0 {
poolLogger.Debugf("add head block [%s] for peer %s", name(hash), peerId)
peer.currentBlockC <- block
} else {
if entry == nil { if entry == nil {
poolLogger.Warnf("unrequested block [%x] by peer %s", hash, peerId) poolLogger.Warnf("unrequested block [%s] by peer %s", name(hash), peerId)
self.peerError(peerId, ErrUnrequestedBlock, "%x", hash) self.peerError(peerId, ErrUnrequestedBlock, "%x", hash)
}
}
if entry == nil {
return return
} }
@ -443,17 +626,21 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
// check if block already present // check if block already present
if node.block != nil { if node.block != nil {
poolLogger.DebugDetailf("block [%x] already sent by %s", name(hash), node.blockBy) poolLogger.DebugDetailf("block [%s] already sent by %s", name(hash), node.blockBy)
return return
} }
if self.hasBlock(hash) {
poolLogger.DebugDetailf("block [%s] already known", name(hash))
} else {
// validate block for PoW // validate block for PoW
if !self.verifyPoW(block) { if !self.verifyPoW(block) {
poolLogger.Warnf("invalid pow on block [%x] by peer %s", hash, peerId) poolLogger.Warnf("invalid pow on block [%s] by peer %s", name(hash), peerId)
self.peerError(peerId, ErrInvalidPoW, "%x", hash) self.peerError(peerId, ErrInvalidPoW, "%x", hash)
return return
} }
}
poolLogger.Debugf("added block [%s] sent by peer %s", name(hash), peerId) poolLogger.Debugf("added block [%s] sent by peer %s", name(hash), peerId)
node.block = block node.block = block
node.blockBy = peerId node.blockBy = peerId
@ -544,23 +731,23 @@ LOOP:
// - when turned off (if peer disconnects and new peer connects with alternative chain), no blockrequests are made but absolute expiry timer is ticking // - when turned off (if peer disconnects and new peer connects with alternative chain), no blockrequests are made but absolute expiry timer is ticking
// - when turned back on it recursively calls itself on the root of the next chain section // - when turned back on it recursively calls itself on the root of the next chain section
// - when exits, signals to // - when exits, signals to
func (self *BlockPool) processSection(section *section, nodes []*poolNode) { func (self *BlockPool) processSection(sec *section, nodes []*poolNode) {
for i, node := range nodes { for i, node := range nodes {
entry := &poolEntry{node: node, section: section, index: i} entry := &poolEntry{node: node, section: sec, index: i}
self.set(node.hash, entry) self.set(node.hash, entry)
} }
section.bottom = nodes[len(nodes)-1] sec.bottom = nodes[len(nodes)-1]
section.top = nodes[0] sec.top = nodes[0]
section.nodes = nodes sec.nodes = nodes
poolLogger.DebugDetailf("[%s] setup section process", sectionName(section)) poolLogger.DebugDetailf("[%s] setup section process", sectionName(sec))
self.wg.Add(1) self.wg.Add(1)
go func() { go func() {
// absolute time after which sub-chain is killed if not complete (some blocks are missing) // absolute time after which sub-chain is killed if not complete (some blocks are missing)
suicideTimer := time.After(blockTimeout * time.Minute) suicideTimer := time.After(blocksTimeout * time.Second)
var peer, newPeer *peerInfo var peer, newPeer *peerInfo
@ -580,21 +767,23 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
var insertChain bool var insertChain bool
var quitC chan bool var quitC chan bool
var blockChainC = section.blockChainC var blockChainC = sec.blockChainC
var parentHash []byte
LOOP: LOOP:
for { for {
if insertChain { if insertChain {
insertChain = false insertChain = false
rest, err := self.addSectionToBlockChain(section) rest, err := self.addSectionToBlockChain(sec)
if err != nil { if err != nil {
close(section.suicideC) close(sec.suicideC)
continue LOOP continue LOOP
} }
if rest == 0 { if rest == 0 {
blocksRequestsComplete = true blocksRequestsComplete = true
child := self.getChild(section) child := self.getChild(sec)
if child != nil { if child != nil {
self.connectToBlockChain(child) self.connectToBlockChain(child)
} }
@ -603,7 +792,7 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
if blockHashesRequestsComplete && blocksRequestsComplete { if blockHashesRequestsComplete && blocksRequestsComplete {
// not waiting for hashes any more // not waiting for hashes any more
poolLogger.Debugf("[%s] section complete %v blocks retrieved (%v attempts), hash requests complete on root (%v attempts)", sectionName(section), depth, blocksRequests, blockHashesRequests) poolLogger.Debugf("[%s] section complete %v blocks retrieved (%v attempts), hash requests complete on root (%v attempts)", sectionName(sec), depth, blocksRequests, blockHashesRequests)
break LOOP break LOOP
} // otherwise suicide if no hashes coming } // otherwise suicide if no hashes coming
@ -611,11 +800,12 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
// went through all blocks in section // went through all blocks in section
if missing == 0 { if missing == 0 {
// no missing blocks // no missing blocks
poolLogger.DebugDetailf("[%s] got all blocks. process complete (%v total blocksRequests): missing %v/%v/%v", sectionName(section), blocksRequests, missing, lastMissing, depth) poolLogger.DebugDetailf("[%s] got all blocks. process complete (%v total blocksRequests): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
blocksRequestsComplete = true blocksRequestsComplete = true
blocksRequestTimer = nil blocksRequestTimer = nil
blocksRequestTime = false blocksRequestTime = false
} else { } else {
poolLogger.DebugDetailf("[%s] section checked: missing %v/%v/%v", sectionName(sec), missing, lastMissing, depth)
// some missing blocks // some missing blocks
blocksRequests++ blocksRequests++
if len(hashes) > 0 { if len(hashes) > 0 {
@ -630,8 +820,8 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
idle++ idle++
// too many idle rounds // too many idle rounds
if idle >= blocksRequestMaxIdleRounds { if idle >= blocksRequestMaxIdleRounds {
poolLogger.DebugDetailf("[%s] block requests had %v idle rounds (%v total attempts): missing %v/%v/%v\ngiving up...", sectionName(section), idle, blocksRequests, missing, lastMissing, depth) poolLogger.DebugDetailf("[%s] block requests had %v idle rounds (%v total attempts): missing %v/%v/%v\ngiving up...", sectionName(sec), idle, blocksRequests, missing, lastMissing, depth)
close(section.suicideC) close(sec.suicideC)
} }
} else { } else {
idle = 0 idle = 0
@ -653,22 +843,39 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
// //
if ready && blocksRequestTime && !blocksRequestsComplete { if ready && blocksRequestTime && !blocksRequestsComplete {
poolLogger.DebugDetailf("[%s] check if new blocks arrived (attempt %v): missing %v/%v/%v", sectionName(section), blocksRequests, missing, lastMissing, depth) poolLogger.DebugDetailf("[%s] check if new blocks arrived (attempt %v): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
blocksRequestTimer = time.After(blocksRequestInterval * time.Millisecond) blocksRequestTimer = time.After(blocksRequestInterval * time.Millisecond)
blocksRequestTime = false blocksRequestTime = false
processC = offC processC = offC
} }
if blockHashesRequestTime { if blockHashesRequestTime {
if self.getParent(section) != nil { var parentSection = self.getParent(sec)
if parentSection == nil {
if parent := self.get(parentHash); parent != nil {
parentSection = parent.section
self.chainLock.Lock()
self.link(parentSection, sec)
self.chainLock.Unlock()
} else {
if self.hasBlock(parentHash) {
insertChain = true
blockHashesRequestTime = false
blockHashesRequestTimer = nil
blockHashesRequestsComplete = true
continue LOOP
}
}
}
if parentSection != nil {
// if not root of chain, switch off // if not root of chain, switch off
poolLogger.DebugDetailf("[%s] parent found, hash requests deactivated (after %v total attempts)\n", sectionName(section), blockHashesRequests) poolLogger.DebugDetailf("[%s] parent found, hash requests deactivated (after %v total attempts)\n", sectionName(sec), blockHashesRequests)
blockHashesRequestTimer = nil blockHashesRequestTimer = nil
blockHashesRequestsComplete = true blockHashesRequestsComplete = true
} else { } else {
blockHashesRequests++ blockHashesRequests++
poolLogger.Debugf("[%s] hash request on root (%v total attempts)\n", sectionName(section), blockHashesRequests) poolLogger.Debugf("[%s] hash request on root (%v total attempts)\n", sectionName(sec), blockHashesRequests)
peer.requestBlockHashes(section.bottom.hash) peer.requestBlockHashes(sec.bottom.hash)
blockHashesRequestTimer = time.After(blockHashesRequestInterval * time.Millisecond) blockHashesRequestTimer = time.After(blockHashesRequestInterval * time.Millisecond)
} }
blockHashesRequestTime = false blockHashesRequestTime = false
@ -682,27 +889,27 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
// peer quit or demoted, put section in idle mode // peer quit or demoted, put section in idle mode
quitC = nil quitC = nil
go func() { go func() {
section.controlC <- nil sec.controlC <- nil
}() }()
case <-self.purgeC: case <-self.purgeC:
suicideTimer = time.After(0) suicideTimer = time.After(0)
case <-suicideTimer: case <-suicideTimer:
close(section.suicideC) close(sec.suicideC)
poolLogger.Debugf("[%s] timeout. (%v total attempts): missing %v/%v/%v", sectionName(section), blocksRequests, missing, lastMissing, depth) poolLogger.Debugf("[%s] timeout. (%v total attempts): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
case <-section.suicideC: case <-sec.suicideC:
poolLogger.Debugf("[%s] suicide", sectionName(section)) poolLogger.Debugf("[%s] suicide", sectionName(sec))
// first delink from child and parent under chainlock // first delink from child and parent under chainlock
self.chainLock.Lock() self.chainLock.Lock()
self.link(nil, section) self.link(nil, sec)
self.link(section, nil) self.link(sec, nil)
self.chainLock.Unlock() self.chainLock.Unlock()
// delete node entries from pool index under pool lock // delete node entries from pool index under pool lock
self.lock.Lock() self.lock.Lock()
for _, node := range section.nodes { for _, node := range sec.nodes {
delete(self.pool, string(node.hash)) delete(self.pool, string(node.hash))
} }
self.lock.Unlock() self.lock.Unlock()
@ -710,20 +917,20 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
break LOOP break LOOP
case <-blocksRequestTimer: case <-blocksRequestTimer:
poolLogger.DebugDetailf("[%s] block request time", sectionName(section)) poolLogger.DebugDetailf("[%s] block request time", sectionName(sec))
blocksRequestTime = true blocksRequestTime = true
case <-blockHashesRequestTimer: case <-blockHashesRequestTimer:
poolLogger.DebugDetailf("[%s] hash request time", sectionName(section)) poolLogger.DebugDetailf("[%s] hash request time", sectionName(sec))
blockHashesRequestTime = true blockHashesRequestTime = true
case newPeer = <-section.controlC: case newPeer = <-sec.controlC:
// active -> idle // active -> idle
if peer != nil && newPeer == nil { if peer != nil && newPeer == nil {
self.procWg.Done() self.procWg.Done()
if init { if init {
poolLogger.Debugf("[%s] idle mode (%v total attempts): missing %v/%v/%v", sectionName(section), blocksRequests, missing, lastMissing, depth) poolLogger.Debugf("[%s] idle mode (%v total attempts): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
} }
blocksRequestTime = false blocksRequestTime = false
blocksRequestTimer = nil blocksRequestTimer = nil
@ -739,11 +946,11 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
if peer == nil && newPeer != nil { if peer == nil && newPeer != nil {
self.procWg.Add(1) self.procWg.Add(1)
poolLogger.Debugf("[%s] active mode", sectionName(section)) poolLogger.Debugf("[%s] active mode", sectionName(sec))
if !blocksRequestsComplete { if !blocksRequestsComplete {
blocksRequestTime = true blocksRequestTime = true
} }
if !blockHashesRequestsComplete { if !blockHashesRequestsComplete && parentHash != nil {
blockHashesRequestTime = true blockHashesRequestTime = true
} }
if !init { if !init {
@ -753,13 +960,13 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
missing = 0 missing = 0
self.wg.Add(1) self.wg.Add(1)
self.procWg.Add(1) self.procWg.Add(1)
depth = len(section.nodes) depth = len(sec.nodes)
lastMissing = depth lastMissing = depth
// if not run at least once fully, launch iterator // if not run at least once fully, launch iterator
go func() { go func() {
var node *poolNode var node *poolNode
IT: IT:
for _, node = range section.nodes { for _, node = range sec.nodes {
select { select {
case processC <- node: case processC <- node:
case <-self.quit: case <-self.quit:
@ -771,7 +978,7 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
self.procWg.Done() self.procWg.Done()
}() }()
} else { } else {
poolLogger.Debugf("[%s] restore earlier state", sectionName(section)) poolLogger.Debugf("[%s] restore earlier state", sectionName(sec))
processC = offC processC = offC
} }
} }
@ -781,7 +988,7 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
} }
peer = newPeer peer = newPeer
case waiter := <-section.forkC: case waiter := <-sec.forkC:
// this case just blocks the process until section is split at the fork // this case just blocks the process until section is split at the fork
<-waiter <-waiter
init = false init = false
@ -794,7 +1001,7 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
init = true init = true
done = true done = true
processC = make(chan *poolNode, missing) processC = make(chan *poolNode, missing)
poolLogger.DebugDetailf("[%s] section initalised: missing %v/%v/%v", sectionName(section), missing, lastMissing, depth) poolLogger.DebugDetailf("[%s] section initalised: missing %v/%v/%v", sectionName(sec), missing, lastMissing, depth)
continue LOOP continue LOOP
} }
if ready { if ready {
@ -811,17 +1018,24 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
missing++ missing++
hashes = append(hashes, node.hash) hashes = append(hashes, node.hash)
if len(hashes) == blockBatchSize { if len(hashes) == blockBatchSize {
poolLogger.Debugf("[%s] request %v missing blocks", sectionName(section), len(hashes)) poolLogger.Debugf("[%s] request %v missing blocks", sectionName(sec), len(hashes))
self.requestBlocks(blocksRequests, hashes) self.requestBlocks(blocksRequests, hashes)
hashes = nil hashes = nil
} }
missingC <- node missingC <- node
} else { } else {
if blockChainC == nil && i == lastMissing { if i == lastMissing {
if blockChainC == nil {
insertChain = true insertChain = true
} else {
if parentHash == nil {
parentHash = block.ParentHash()
poolLogger.Debugf("[%s] found root block [%s]", sectionName(sec), name(parentHash))
blockHashesRequestTime = true
}
}
} }
} }
poolLogger.Debugf("[%s] %v/%v/%v/%v", sectionName(section), i, missing, lastMissing, depth)
if i == lastMissing && init { if i == lastMissing && init {
done = true done = true
} }
@ -829,23 +1043,22 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
case <-blockChainC: case <-blockChainC:
// closed blockChain channel indicates that the blockpool is reached // closed blockChain channel indicates that the blockpool is reached
// connected to the blockchain, insert the longest chain of blocks // connected to the blockchain, insert the longest chain of blocks
poolLogger.Debugf("[%s] reached blockchain", sectionName(section)) poolLogger.Debugf("[%s] reached blockchain", sectionName(sec))
blockChainC = nil blockChainC = nil
// switch off hash requests in case they were on // switch off hash requests in case they were on
blockHashesRequestTime = false blockHashesRequestTime = false
blockHashesRequestTimer = nil blockHashesRequestTimer = nil
blockHashesRequestsComplete = true blockHashesRequestsComplete = true
// section root has block // section root has block
if len(section.nodes) > 0 && section.nodes[len(section.nodes)-1].block != nil { if len(sec.nodes) > 0 && sec.nodes[len(sec.nodes)-1].block != nil {
insertChain = true insertChain = true
} }
continue LOOP continue LOOP
} // select } // select
} // for } // for
poolLogger.Debugf("[%s] section complete: %v block hashes requests - %v block requests - missing %v/%v/%v", sectionName(section), blockHashesRequests, blocksRequests, missing, lastMissing, depth)
close(section.offC) close(sec.offC)
self.wg.Done() self.wg.Done()
if peer != nil { if peer != nil {
@ -917,22 +1130,28 @@ func (self *peerInfo) addSection(hash []byte, section *section) (found *section)
defer self.lock.Unlock() defer self.lock.Unlock()
key := string(hash) key := string(hash)
found = self.sections[key] found = self.sections[key]
poolLogger.DebugDetailf("[%s] section process %s registered", sectionName(section), self.id) poolLogger.DebugDetailf("[%s] section process stored for %s", sectionName(section), self.id)
self.sections[key] = section self.sections[key] = section
return return
} }
func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) { func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) {
if newPeer != nil { if newPeer != nil {
entry := self.get(newPeer.currentBlock) newPeer.quitC = make(chan bool)
if entry == nil {
poolLogger.Debugf("[%s] head block [%s] not found, requesting hashes", newPeer.id, name(newPeer.currentBlock))
newPeer.requestBlockHashes(newPeer.currentBlock)
} else {
poolLogger.Debugf("[%s] head block [%s] found, activate chain at section [%s]", newPeer.id, name(newPeer.currentBlock), sectionName(entry.section))
self.activateChain(entry.section, newPeer)
}
poolLogger.DebugDetailf("[%s] activate section processes", newPeer.id) poolLogger.DebugDetailf("[%s] activate section processes", newPeer.id)
var addSections []*section
for hash, section := range newPeer.sections {
// split sections get reorganised here
if string(section.top.hash) != hash {
addSections = append(addSections, section)
if entry := self.get([]byte(hash)); entry != nil {
addSections = append(addSections, entry.section)
}
}
}
for _, section := range addSections {
newPeer.sections[string(section.top.hash)] = section
}
for hash, section := range newPeer.sections { for hash, section := range newPeer.sections {
// this will block if section process is waiting for peer lock // this will block if section process is waiting for peer lock
select { select {
@ -940,12 +1159,26 @@ func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) {
poolLogger.DebugDetailf("[%s][%x] section process complete - remove", newPeer.id, hash[:4]) poolLogger.DebugDetailf("[%s][%x] section process complete - remove", newPeer.id, hash[:4])
delete(newPeer.sections, hash) delete(newPeer.sections, hash)
case section.controlC <- newPeer: case section.controlC <- newPeer:
poolLogger.DebugDetailf("[%s][%x] registered peer with section", newPeer.id, hash[:4]) poolLogger.DebugDetailf("[%s][%x] activates section [%s]", newPeer.id, hash[:4], sectionName(section))
} }
} }
newPeer.quitC = make(chan bool) newPeer.lock.Lock()
headSection := newPeer.headSection
currentBlockHash := newPeer.currentBlockHash
newPeer.lock.Unlock()
if headSection == nil {
poolLogger.DebugDetailf("[%s] head section for [%s] not created, requesting info", newPeer.id, name(currentBlockHash))
self.requestHeadSection(newPeer)
} else {
if entry := self.get(currentBlockHash); entry != nil {
headSection = entry.section
}
poolLogger.DebugDetailf("[%s] activate chain at head section [%s] for current head [%s]", newPeer.id, sectionName(headSection), name(currentBlockHash))
self.activateChain(headSection, newPeer)
}
} }
if oldPeer != nil { if oldPeer != nil {
poolLogger.DebugDetailf("[%s] quit section processes", oldPeer.id)
close(oldPeer.quitC) close(oldPeer.quitC)
} }
} }

188
eth/block_pool_test.go
Unescape View File

@ -18,7 +18,7 @@ import (
const waitTimeout = 60 // seconds const waitTimeout = 60 // seconds
var logsys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugLevel)) var logsys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel))
var ini = false var ini = false
@ -336,12 +336,12 @@ func (self *peerTester) AddPeer() bool {
// peer sends blockhashes if and when gets a request // peer sends blockhashes if and when gets a request
func (self *peerTester) AddBlockHashes(indexes ...int) { func (self *peerTester) AddBlockHashes(indexes ...int) {
i := 0
fmt.Printf("ready to add block hashes %v\n", indexes) fmt.Printf("ready to add block hashes %v\n", indexes)
self.waitBlockHashesRequests(indexes[0]) self.waitBlockHashesRequests(indexes[0])
fmt.Printf("adding block hashes %v\n", indexes) fmt.Printf("adding block hashes %v\n", indexes)
hashes := self.hashPool.indexesToHashes(indexes) hashes := self.hashPool.indexesToHashes(indexes)
i := 1
next := func() (hash []byte, ok bool) { next := func() (hash []byte, ok bool) {
if i < len(hashes) { if i < len(hashes) {
hash = hashes[i] hash = hashes[i]
@ -415,7 +415,7 @@ func TestAddPeer(t *testing.T) {
if blockPool.peer.id != "peer0" { if blockPool.peer.id != "peer0" {
t.Errorf("peer0 (TD=1) not set as best") t.Errorf("peer0 (TD=1) not set as best")
} }
peer0.checkBlockHashesRequests(0) // peer0.checkBlockHashesRequests(0)
best = peer2.AddPeer() best = peer2.AddPeer()
if !best { if !best {
@ -424,7 +424,7 @@ func TestAddPeer(t *testing.T) {
if blockPool.peer.id != "peer2" { if blockPool.peer.id != "peer2" {
t.Errorf("peer2 (TD=3) not set as best") t.Errorf("peer2 (TD=3) not set as best")
} }
peer2.checkBlockHashesRequests(2) peer2.waitBlocksRequests(2)
best = peer1.AddPeer() best = peer1.AddPeer()
if best { if best {
@ -449,7 +449,7 @@ func TestAddPeer(t *testing.T) {
if blockPool.peer.td.Cmp(big.NewInt(int64(4))) != 0 { if blockPool.peer.td.Cmp(big.NewInt(int64(4))) != 0 {
t.Errorf("peer2 TD not updated") t.Errorf("peer2 TD not updated")
} }
peer2.checkBlockHashesRequests(2, 3) peer2.waitBlocksRequests(3)
peer1.td = 3 peer1.td = 3
peer1.currentBlock = 2 peer1.currentBlock = 2
@ -474,7 +474,7 @@ func TestAddPeer(t *testing.T) {
if blockPool.peer.id != "peer1" { if blockPool.peer.id != "peer1" {
t.Errorf("existing peer1 (TD=3) should be set as best peer") t.Errorf("existing peer1 (TD=3) should be set as best peer")
} }
peer1.checkBlockHashesRequests(2) peer1.waitBlocksRequests(2)
blockPool.RemovePeer("peer1") blockPool.RemovePeer("peer1")
peer, best = blockPool.getPeer("peer1") peer, best = blockPool.getPeer("peer1")
@ -485,6 +485,7 @@ func TestAddPeer(t *testing.T) {
if blockPool.peer.id != "peer0" { if blockPool.peer.id != "peer0" {
t.Errorf("existing peer0 (TD=1) should be set as best peer") t.Errorf("existing peer0 (TD=1) should be set as best peer")
} }
peer0.waitBlocksRequests(0)
blockPool.RemovePeer("peer0") blockPool.RemovePeer("peer0")
peer, best = blockPool.getPeer("peer0") peer, best = blockPool.getPeer("peer0")
@ -502,7 +503,7 @@ func TestAddPeer(t *testing.T) {
if blockPool.peer.id != "peer0" { if blockPool.peer.id != "peer0" {
t.Errorf("peer0 (TD=1) should be set as best") t.Errorf("peer0 (TD=1) should be set as best")
} }
peer0.checkBlockHashesRequests(0, 0, 3) peer0.waitBlocksRequests(3)
blockPool.Stop() blockPool.Stop()
@ -513,17 +514,36 @@ func TestPeerWithKnownBlock(t *testing.T) {
_, blockPool, blockPoolTester := newTestBlockPool(t) _, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.refBlockChain[0] = nil blockPoolTester.refBlockChain[0] = nil
blockPoolTester.blockChain[0] = nil blockPoolTester.blockChain[0] = nil
// hashPool, blockPool, blockPoolTester := newTestBlockPool()
blockPool.Start() blockPool.Start()
peer0 := blockPoolTester.newPeer("0", 1, 0) peer0 := blockPoolTester.newPeer("0", 1, 0)
peer0.AddPeer() peer0.AddPeer()
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop() blockPool.Stop()
// no request on known block // no request on known block
peer0.checkBlockHashesRequests() peer0.checkBlockHashesRequests()
} }
func TestPeerWithKnownParentBlock(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.initRefBlockChain(1)
blockPoolTester.blockChain[0] = nil
blockPool.Start()
peer0 := blockPoolTester.newPeer("0", 1, 1)
peer0.AddPeer()
peer0.AddBlocks(0, 1)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
peer0.checkBlocksRequests([]int{1})
peer0.checkBlockHashesRequests()
blockPoolTester.refBlockChain[1] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestSimpleChain(t *testing.T) { func TestSimpleChain(t *testing.T) {
logInit() logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t) _, blockPool, blockPoolTester := newTestBlockPool(t)
@ -534,8 +554,9 @@ func TestSimpleChain(t *testing.T) {
peer1 := blockPoolTester.newPeer("peer1", 1, 2) peer1 := blockPoolTester.newPeer("peer1", 1, 2)
peer1.AddPeer() peer1.AddPeer()
peer1.AddBlocks(1, 2)
go peer1.AddBlockHashes(2, 1, 0) go peer1.AddBlockHashes(2, 1, 0)
peer1.AddBlocks(0, 1, 2) peer1.AddBlocks(0, 1)
blockPool.Wait(waitTimeout * time.Second) blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop() blockPool.Stop()
@ -543,6 +564,26 @@ func TestSimpleChain(t *testing.T) {
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain) blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
} }
func TestChainConnectingWithParentHash(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(3)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 3)
peer1.AddPeer()
go peer1.AddBlocks(2, 3)
go peer1.AddBlockHashes(3, 2, 1)
peer1.AddBlocks(0, 1, 2)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[3] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestInvalidBlock(t *testing.T) { func TestInvalidBlock(t *testing.T) {
logInit() logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t) _, blockPool, blockPoolTester := newTestBlockPool(t)
@ -554,8 +595,9 @@ func TestInvalidBlock(t *testing.T) {
peer1 := blockPoolTester.newPeer("peer1", 1, 3) peer1 := blockPoolTester.newPeer("peer1", 1, 3)
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlocks(2, 3)
go peer1.AddBlockHashes(3, 2, 1, 0) go peer1.AddBlockHashes(3, 2, 1, 0)
peer1.AddBlocks(0, 1, 2, 3) peer1.AddBlocks(0, 1, 2)
blockPool.Wait(waitTimeout * time.Second) blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop() blockPool.Stop()
@ -566,7 +608,7 @@ func TestInvalidBlock(t *testing.T) {
t.Errorf("wrong error, got %v, expected %v", peer1.peerErrors[0], ErrInvalidBlock) t.Errorf("wrong error, got %v, expected %v", peer1.peerErrors[0], ErrInvalidBlock)
} }
} else { } else {
t.Errorf("expected invalid block error, got nothing") t.Errorf("expected invalid block error, got nothing %v", peer1.peerErrors)
} }
} }
@ -579,7 +621,7 @@ func TestVerifyPoW(t *testing.T) {
blockPoolTester.blockPool.verifyPoW = func(b pow.Block) bool { blockPoolTester.blockPool.verifyPoW = func(b pow.Block) bool {
bb, _ := b.(*types.Block) bb, _ := b.(*types.Block)
indexes := blockPoolTester.hashPool.hashesToIndexes([][]byte{bb.Hash()}) indexes := blockPoolTester.hashPool.hashesToIndexes([][]byte{bb.Hash()})
if indexes[0] == 1 && !first { if indexes[0] == 2 && !first {
first = true first = true
return false return false
} else { } else {
@ -590,15 +632,17 @@ func TestVerifyPoW(t *testing.T) {
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 2) peer1 := blockPoolTester.newPeer("peer1", 1, 3)
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlockHashes(2, 1, 0) go peer1.AddBlocks(2, 3)
go peer1.AddBlockHashes(3, 2, 1, 0)
peer1.AddBlocks(0, 1, 2) peer1.AddBlocks(0, 1, 2)
peer1.AddBlocks(0, 1)
blockPool.Wait(waitTimeout * time.Second) // blockPool.Wait(waitTimeout * time.Second)
time.Sleep(1 * time.Second)
blockPool.Stop() blockPool.Stop()
blockPoolTester.refBlockChain[2] = []int{} blockPoolTester.refBlockChain[1] = []int{}
delete(blockPoolTester.refBlockChain, 2)
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain) blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
if len(peer1.peerErrors) == 1 { if len(peer1.peerErrors) == 1 {
if peer1.peerErrors[0] != ErrInvalidPoW { if peer1.peerErrors[0] != ErrInvalidPoW {
@ -620,8 +664,9 @@ func TestMultiSectionChain(t *testing.T) {
peer1 := blockPoolTester.newPeer("peer1", 1, 5) peer1 := blockPoolTester.newPeer("peer1", 1, 5)
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlocks(4, 5)
go peer1.AddBlockHashes(5, 4, 3) go peer1.AddBlockHashes(5, 4, 3)
go peer1.AddBlocks(2, 3, 4, 5) go peer1.AddBlocks(2, 3, 4)
go peer1.AddBlockHashes(3, 2, 1, 0) go peer1.AddBlockHashes(3, 2, 1, 0)
peer1.AddBlocks(0, 1, 2) peer1.AddBlocks(0, 1, 2)
@ -641,14 +686,17 @@ func TestNewBlocksOnPartialChain(t *testing.T) {
peer1 := blockPoolTester.newPeer("peer1", 1, 5) peer1 := blockPoolTester.newPeer("peer1", 1, 5)
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlocks(4, 5) // partially complete section
go peer1.AddBlockHashes(5, 4, 3) go peer1.AddBlockHashes(5, 4, 3)
peer1.AddBlocks(2, 3) // partially complete section peer1.AddBlocks(3, 4) // partially complete section
// peer1 found new blocks // peer1 found new blocks
peer1.td = 2 peer1.td = 2
peer1.currentBlock = 7 peer1.currentBlock = 7
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlocks(6, 7)
go peer1.AddBlockHashes(7, 6, 5) go peer1.AddBlockHashes(7, 6, 5)
go peer1.AddBlocks(3, 4, 5, 6, 7) go peer1.AddBlocks(2, 3)
go peer1.AddBlocks(5, 6)
go peer1.AddBlockHashes(3, 2, 1, 0) // tests that hash request from known chain root is remembered go peer1.AddBlockHashes(3, 2, 1, 0) // tests that hash request from known chain root is remembered
peer1.AddBlocks(0, 1, 2) peer1.AddBlocks(0, 1, 2)
@ -658,35 +706,37 @@ func TestNewBlocksOnPartialChain(t *testing.T) {
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain) blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
} }
func TestPeerSwitch(t *testing.T) { func TestPeerSwitchUp(t *testing.T) {
logInit() logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t) _, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(6) blockPoolTester.initRefBlockChain(7)
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 5) peer1 := blockPoolTester.newPeer("peer1", 1, 6)
peer2 := blockPoolTester.newPeer("peer2", 2, 6) peer2 := blockPoolTester.newPeer("peer2", 2, 7)
peer2.blocksRequestsMap = peer1.blocksRequestsMap peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlockHashes(5, 4, 3) go peer1.AddBlocks(5, 6)
go peer1.AddBlockHashes(6, 5, 4, 3) //
peer1.AddBlocks(2, 3) // section partially complete, block 3 will be preserved after peer demoted peer1.AddBlocks(2, 3) // section partially complete, block 3 will be preserved after peer demoted
peer2.AddPeer() // peer2 is promoted as best peer, peer1 is demoted peer2.AddPeer() // peer2 is promoted as best peer, peer1 is demoted
go peer2.AddBlockHashes(6, 5) // go peer2.AddBlocks(6, 7)
go peer2.AddBlocks(4, 5, 6) // tests that block request for earlier section is remembered go peer2.AddBlockHashes(7, 6) //
go peer2.AddBlocks(4, 5) // tests that block request for earlier section is remembered
go peer1.AddBlocks(3, 4) // tests that connecting section by demoted peer is remembered and blocks are accepted from demoted peer go peer1.AddBlocks(3, 4) // tests that connecting section by demoted peer is remembered and blocks are accepted from demoted peer
go peer2.AddBlockHashes(3, 2, 1, 0) // tests that known chain section is activated, hash requests from 3 is remembered go peer2.AddBlockHashes(3, 2, 1, 0) // tests that known chain section is activated, hash requests from 3 is remembered
peer2.AddBlocks(0, 1, 2) // final blocks linking to blockchain sent peer2.AddBlocks(0, 1, 2) // final blocks linking to blockchain sent
blockPool.Wait(waitTimeout * time.Second) blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop() blockPool.Stop()
blockPoolTester.refBlockChain[6] = []int{} blockPoolTester.refBlockChain[7] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain) blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
} }
func TestPeerDownSwitch(t *testing.T) { func TestPeerSwitchDown(t *testing.T) {
logInit() logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t) _, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil blockPoolTester.blockChain[0] = nil
@ -698,12 +748,39 @@ func TestPeerDownSwitch(t *testing.T) {
peer2.blocksRequestsMap = peer1.blocksRequestsMap peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer2.AddPeer() peer2.AddPeer()
go peer2.AddBlockHashes(6, 5, 4)
peer2.AddBlocks(5, 6) // partially complete, section will be preserved peer2.AddBlocks(5, 6) // partially complete, section will be preserved
go peer2.AddBlockHashes(6, 5, 4) //
peer2.AddBlocks(4, 5) //
blockPool.RemovePeer("peer2") // peer2 disconnects blockPool.RemovePeer("peer2") // peer2 disconnects
peer1.AddPeer() // inferior peer1 is promoted as best peer peer1.AddPeer() // inferior peer1 is promoted as best peer
go peer1.AddBlockHashes(4, 3, 2, 1, 0) // go peer1.AddBlockHashes(4, 3, 2, 1, 0) //
go peer1.AddBlocks(3, 4, 5) // tests that section set by demoted peer is remembered and blocks are accepted go peer1.AddBlocks(3, 4) // tests that section set by demoted peer is remembered and blocks are accepted , this connects the chain sections together
peer1.AddBlocks(0, 1, 2, 3)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[6] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestPeerCompleteSectionSwitchDown(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(6)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 4)
peer2 := blockPoolTester.newPeer("peer2", 2, 6)
peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer2.AddPeer()
peer2.AddBlocks(5, 6) // partially complete, section will be preserved
go peer2.AddBlockHashes(6, 5, 4) //
peer2.AddBlocks(3, 4, 5) // complete section
blockPool.RemovePeer("peer2") // peer2 disconnects
peer1.AddPeer() // inferior peer1 is promoted as best peer
peer1.AddBlockHashes(4, 3, 2, 1, 0) // tests that hash request are directly connecting if the head block exists
peer1.AddBlocks(0, 1, 2, 3) peer1.AddBlocks(0, 1, 2, 3)
blockPool.Wait(waitTimeout * time.Second) blockPool.Wait(waitTimeout * time.Second)
@ -725,11 +802,13 @@ func TestPeerSwitchBack(t *testing.T) {
peer2.blocksRequestsMap = peer1.blocksRequestsMap peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer2.AddPeer() peer2.AddPeer()
go peer2.AddBlocks(7, 8)
go peer2.AddBlockHashes(8, 7, 6) go peer2.AddBlockHashes(8, 7, 6)
go peer2.AddBlockHashes(6, 5, 4) go peer2.AddBlockHashes(6, 5, 4)
peer2.AddBlocks(4, 5) // section partially complete peer2.AddBlocks(4, 5) // section partially complete
peer1.AddPeer() // peer1 is promoted as best peer peer1.AddPeer() // peer1 is promoted as best peer
go peer1.AddBlockHashes(11, 10) // only gives useless results go peer1.AddBlocks(10, 11) //
peer1.AddBlockHashes(11, 10) // only gives useless results
blockPool.RemovePeer("peer1") // peer1 disconnects blockPool.RemovePeer("peer1") // peer1 disconnects
go peer2.AddBlockHashes(4, 3, 2, 1, 0) // tests that asking for hashes from 4 is remembered go peer2.AddBlockHashes(4, 3, 2, 1, 0) // tests that asking for hashes from 4 is remembered
go peer2.AddBlocks(3, 4, 5, 6, 7, 8) // tests that section 4, 5, 6 and 7, 8 are remembered for missing blocks go peer2.AddBlocks(3, 4, 5, 6, 7, 8) // tests that section 4, 5, 6 and 7, 8 are remembered for missing blocks
@ -756,11 +835,13 @@ func TestForkSimple(t *testing.T) {
peer2.blocksRequestsMap = peer1.blocksRequestsMap peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlocks(8, 9)
go peer1.AddBlockHashes(9, 8, 7, 3, 2) go peer1.AddBlockHashes(9, 8, 7, 3, 2)
peer1.AddBlocks(1, 2, 3, 7, 8, 9) peer1.AddBlocks(1, 2, 3, 7, 8)
peer2.AddPeer() // peer2 is promoted as best peer peer2.AddPeer() // peer2 is promoted as best peer
go peer2.AddBlocks(5, 6) //
go peer2.AddBlockHashes(6, 5, 4, 3, 2) // fork on 3 -> 4 (earlier child: 7) go peer2.AddBlockHashes(6, 5, 4, 3, 2) // fork on 3 -> 4 (earlier child: 7)
go peer2.AddBlocks(1, 2, 3, 4, 5, 6) go peer2.AddBlocks(1, 2, 3, 4, 5)
go peer2.AddBlockHashes(2, 1, 0) go peer2.AddBlockHashes(2, 1, 0)
peer2.AddBlocks(0, 1, 2) peer2.AddBlocks(0, 1, 2)
@ -790,23 +871,24 @@ func TestForkSwitchBackByNewBlocks(t *testing.T) {
peer2.blocksRequestsMap = peer1.blocksRequestsMap peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlockHashes(9, 8, 7, 3, 2) peer1.AddBlocks(8, 9) //
peer1.AddBlocks(8, 9) // partial section go peer1.AddBlockHashes(9, 8, 7, 3, 2) //
peer1.AddBlocks(7, 8) // partial section
peer2.AddPeer() // peer2.AddPeer() //
peer2.AddBlocks(5, 6) //
go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3 go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
peer2.AddBlocks(1, 2, 3, 4, 5, 6) // peer2.AddBlocks(1, 2, 3, 4, 5) //
// peer1 finds new blocks // peer1 finds new blocks
peer1.td = 3 peer1.td = 3
peer1.currentBlock = 11 peer1.currentBlock = 11
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlocks(10, 11)
go peer1.AddBlockHashes(11, 10, 9) go peer1.AddBlockHashes(11, 10, 9)
peer1.AddBlocks(7, 8, 9, 10, 11) peer1.AddBlocks(9, 10)
go peer1.AddBlockHashes(7, 3) // tests that hash request from fork root is remembered
go peer1.AddBlocks(3, 7) // tests that block requests on earlier fork are remembered go peer1.AddBlocks(3, 7) // tests that block requests on earlier fork are remembered
// go peer1.AddBlockHashes(1, 0) // tests that hash request from root of connecting chain section (added by demoted peer) is remembered
go peer1.AddBlockHashes(2, 1, 0) // tests that hash request from root of connecting chain section (added by demoted peer) is remembered go peer1.AddBlockHashes(2, 1, 0) // tests that hash request from root of connecting chain section (added by demoted peer) is remembered
peer1.AddBlocks(0, 1, 2, 3) peer1.AddBlocks(0, 1)
blockPool.Wait(waitTimeout * time.Second) blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop() blockPool.Stop()
@ -834,16 +916,18 @@ func TestForkSwitchBackByPeerSwitchBack(t *testing.T) {
peer2.blocksRequestsMap = peer1.blocksRequestsMap peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlocks(8, 9)
go peer1.AddBlockHashes(9, 8, 7, 3, 2) go peer1.AddBlockHashes(9, 8, 7, 3, 2)
peer1.AddBlocks(8, 9) peer1.AddBlocks(7, 8)
peer2.AddPeer() // peer2.AddPeer()
go peer2.AddBlocks(5, 6) //
go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3 go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
peer2.AddBlocks(2, 3, 4, 5, 6) // peer2.AddBlocks(2, 3, 4, 5) //
blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer
peer1.AddBlockHashes(7, 3) // tests that hash request from fork root is remembered go peer1.AddBlocks(3, 7) // tests that block requests on earlier fork are remembered and orphan section relinks to existing parent block
go peer1.AddBlocks(3, 7, 8) // tests that block requests on earlier fork are remembered go peer1.AddBlocks(1, 2) //
go peer1.AddBlockHashes(2, 1, 0) // go peer1.AddBlockHashes(2, 1, 0) //
peer1.AddBlocks(0, 1, 2, 3) peer1.AddBlocks(0, 1)
blockPool.Wait(waitTimeout * time.Second) blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop() blockPool.Stop()
@ -871,16 +955,18 @@ func TestForkCompleteSectionSwitchBackByPeerSwitchBack(t *testing.T) {
peer2.blocksRequestsMap = peer1.blocksRequestsMap peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer1.AddPeer() peer1.AddPeer()
go peer1.AddBlocks(8, 9)
go peer1.AddBlockHashes(9, 8, 7) go peer1.AddBlockHashes(9, 8, 7)
peer1.AddBlocks(3, 7, 8, 9) // make sure this section is complete peer1.AddBlocks(3, 7, 8) // make sure this section is complete
time.Sleep(1 * time.Second) time.Sleep(1 * time.Second)
go peer1.AddBlockHashes(7, 3, 2) // block 3/7 is section boundary go peer1.AddBlockHashes(7, 3, 2) // block 3/7 is section boundary
peer1.AddBlocks(2, 3) // partially complete sections peer1.AddBlocks(2, 3) // partially complete sections block 2 missing
peer2.AddPeer() // peer2.AddPeer() //
go peer2.AddBlocks(5, 6) //
go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3 go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
peer2.AddBlocks(2, 3, 4, 5, 6) // block 2 still missing. peer2.AddBlocks(2, 3, 4, 5) // block 2 still missing.
blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer
peer1.AddBlockHashes(7, 3) // tests that hash request from fork root is remembered even though section process completed // peer1.AddBlockHashes(7, 3) // tests that hash request from fork root is remembered even though section process completed
go peer1.AddBlockHashes(2, 1, 0) // go peer1.AddBlockHashes(2, 1, 0) //
peer1.AddBlocks(0, 1, 2) peer1.AddBlocks(0, 1, 2)

2
eth/error.go
Unescape View File

@ -16,6 +16,7 @@ const (
ErrInvalidBlock ErrInvalidBlock
ErrInvalidPoW ErrInvalidPoW
ErrUnrequestedBlock ErrUnrequestedBlock
ErrInsufficientChainInfo
) )
var errorToString = map[int]string{ var errorToString = map[int]string{
@ -30,6 +31,7 @@ var errorToString = map[int]string{
ErrInvalidBlock: "Invalid block", ErrInvalidBlock: "Invalid block",
ErrInvalidPoW: "Invalid PoW", ErrInvalidPoW: "Invalid PoW",
ErrUnrequestedBlock: "Unrequested block", ErrUnrequestedBlock: "Unrequested block",
ErrInsufficientChainInfo: "Insufficient chain info",
} }
type protocolError struct { type protocolError struct {

10
eth/protocol.go
Unescape View File

@ -211,16 +211,6 @@ func (self *ethProtocol) handle() error {
// uses AddPeer followed by AddHashes, AddBlock only if peer is the best peer // uses AddPeer followed by AddHashes, AddBlock only if peer is the best peer
// (or selected as new best peer) // (or selected as new best peer)
if self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) { if self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) {
called := true
iter := func() ([]byte, bool) {
if called {
called = false
return hash, true
} else {
return nil, false
}
}
self.blockPool.AddBlockHashes(iter, self.id)
self.blockPool.AddBlock(request.Block, self.id) self.blockPool.AddBlock(request.Block, self.id)
} }

6
eth/test/tests/02.sh
Unescape View File

@ -12,8 +12,8 @@ EOF
peer 11 01 peer 11 01
peer 12 02 peer 12 02
P13ID=$PID P12ID=$PID
test_node $NAME "" -loglevel 5 $JSFILE test_node $NAME "" -loglevel 5 $JSFILE
sleep 0.5 sleep 0.3
kill $P13ID kill $P12ID

4
eth/test/tests/03.sh
Unescape View File

@ -1,10 +1,10 @@
#!/bin/bash #!/bin/bash
TIMEOUT=35 TIMEOUT=12
cat >> $JSFILE <<EOF cat >> $JSFILE <<EOF
eth.addPeer("localhost:30311"); eth.addPeer("localhost:30311");
sleep(30000); sleep(10000);
eth.export("$CHAIN_TEST"); eth.export("$CHAIN_TEST");
EOF EOF

6
eth/test/tests/common.sh
Unescape View File

@ -3,14 +3,14 @@
# launched by run.sh # launched by run.sh
function test_node { function test_node {
rm -rf $DIR/$1 rm -rf $DIR/$1
ARGS="-datadir $DIR/$1 -debug debug -seed=false -shh=false -id test$1" ARGS="-datadir $DIR/$1 -debug debug -seed=false -shh=false -id test$1 -port 303$1"
if [ "" != "$2" ]; then if [ "" != "$2" ]; then
chain="chains/$2.chain" chain="chains/$2.chain"
echo "import chain $chain" echo "import chain $chain"
$ETH $ARGS -loglevel 3 -chain $chain | grep CLI |grep import $ETH $ARGS -loglevel 3 -chain $chain | grep CLI |grep import
fi fi
echo "starting test node $1 with extra args ${@:3}" echo "starting test node $1 with args $ARGS ${@:3}"
$ETH $ARGS -port 303$1 ${@:3} & $ETH $ARGS ${@:3} &
PID=$! PID=$!
PIDS="$PIDS $PID" PIDS="$PIDS $PID"
} }

Write Preview
Loading…
Cancel
Save