Skip to content

G
Geth-Modification
Commit 16ecda95 authored by zelig's avatar zelig
Browse files
Options

integrate blockpool into eth 

- remove blockpool code
- remove blockpool integration test (kinda embarrassing)
- remove errors.go
parent cdc86965
Hide whitespace changes
Inline Side-by-side
Showing with 61 additions and 2570 deletions
eth/backend.go
View file @ 16ecda95
......@@ -7,6 +7,7 @@ import (
"path"
"strings"
"github.com/ethereum/go-ethereum/blockpool"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb"
......@@ -117,7 +118,7 @@ type Ethereum struct {
blockProcessor *core.BlockProcessor
txPool *core.TxPool
chainManager *core.ChainManager
blockPool *BlockPool
blockPool *blockpool.BlockPool
whisper *whisper.Whisper
net *p2p.Server
......@@ -185,7 +186,7 @@ func New(config *Config) (*Ethereum, error) {
hasBlock := eth.chainManager.HasBlock
insertChain := eth.chainManager.InsertChain
eth.blockPool = NewBlockPool(hasBlock, insertChain, ezp.Verify)
eth.blockPool = blockpool.New(hasBlock, insertChain, ezp.Verify)
netprv, err := config.nodeKey()
if err != nil {
......@@ -220,7 +221,7 @@ func (s *Ethereum) Name() string { return s.net.Name }
func (s *Ethereum) ChainManager() *core.ChainManager { return s.chainManager }
func (s *Ethereum) BlockProcessor() *core.BlockProcessor { return s.blockProcessor }
func (s *Ethereum) TxPool() *core.TxPool { return s.txPool }
func (s *Ethereum) BlockPool() *BlockPool { return s.blockPool }
func (s *Ethereum) BlockPool() *blockpool.BlockPool { return s.blockPool }
func (s *Ethereum) Whisper() *whisper.Whisper { return s.whisper }
func (s *Ethereum) EventMux() *event.TypeMux { return s.eventMux }
func (s *Ethereum) Db() ethutil.Database { return s.db }
......
eth/block_pool.go deleted 100644 → 0
View file @ cdc86965
package eth
import (
"bytes"
"fmt"
"math"
"math/big"
"math/rand"
"sort"
"sync"
"time"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethutil"
ethlogger "github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/pow"
)
var poolLogger = ethlogger.NewLogger("Blockpool")
const (
blockHashesBatchSize = 256
blockBatchSize = 64
blocksRequestInterval = 500 // ms
blocksRequestRepetition = 1
blockHashesRequestInterval = 500 // ms
blocksRequestMaxIdleRounds = 100
blockHashesTimeout = 60 // seconds
blocksTimeout = 120 // seconds
)
type poolNode struct {
lock sync.RWMutex
hash []byte
td *big.Int
block *types.Block
parent *poolNode
peer string
blockBy string
}
type poolEntry struct {
node *poolNode
section *section
index int
}
type BlockPool struct {
lock sync.RWMutex
chainLock sync.RWMutex
pool map[string]*poolEntry
peersLock sync.RWMutex
peers map[string]*peerInfo
peer *peerInfo
quit chan bool
purgeC chan bool
flushC chan bool
wg sync.WaitGroup
procWg sync.WaitGroup
running bool
// the minimal interface with blockchain
hasBlock func(hash []byte) bool
insertChain func(types.Blocks) error
verifyPoW func(pow.Block) bool
}
type peerInfo struct {
lock sync.RWMutex
td *big.Int
currentBlockHash []byte
currentBlock *types.Block
currentBlockC chan *types.Block
parentHash []byte
headSection *section
headSectionC chan *section
id string
requestBlockHashes func([]byte) error
requestBlocks func([][]byte) error
peerError func(int, string, ...interface{})
sections map[string]*section
quitC chan bool
}
// structure to store long range links on chain to skip along
type section struct {
lock sync.RWMutex
parent *section
child *section
top *poolNode
bottom *poolNode
nodes []*poolNode
controlC chan *peerInfo
suicideC chan bool
blockChainC chan bool
forkC chan chan bool
offC chan bool
}
func NewBlockPool(hasBlock func(hash []byte) bool, insertChain func(types.Blocks) error, verifyPoW func(pow.Block) bool,
) *BlockPool {
return &BlockPool{
hasBlock: hasBlock,
insertChain: insertChain,
verifyPoW: verifyPoW,
}
}
// allows restart
func (self *BlockPool) Start() {
self.lock.Lock()
if self.running {
self.lock.Unlock()
return
}
self.running = true
self.quit = make(chan bool)
self.flushC = make(chan bool)
self.pool = make(map[string]*poolEntry)
self.lock.Unlock()
self.peersLock.Lock()
self.peers = make(map[string]*peerInfo)
self.peersLock.Unlock()
poolLogger.Infoln("Started")
}
func (self *BlockPool) Stop() {
self.lock.Lock()
if !self.running {
self.lock.Unlock()
return
}
self.running = false
self.lock.Unlock()
poolLogger.Infoln("Stopping...")
close(self.quit)
//self.wg.Wait()
self.peersLock.Lock()
self.peers = nil
self.peer = nil
self.peersLock.Unlock()
self.lock.Lock()
self.pool = nil
self.lock.Unlock()
poolLogger.Infoln("Stopped")
}
func (self *BlockPool) Purge() {
self.lock.Lock()
if !self.running {
self.lock.Unlock()
return
}
self.lock.Unlock()
poolLogger.Infoln("Purging...")
close(self.purgeC)
self.wg.Wait()
self.purgeC = make(chan bool)
poolLogger.Infoln("Stopped")
}
func (self *BlockPool) Wait(t time.Duration) {
self.lock.Lock()
if !self.running {
self.lock.Unlock()
return
}
self.lock.Unlock()
poolLogger.Infoln("Waiting for processes to complete...")
close(self.flushC)
w := make(chan bool)
go func() {
self.procWg.Wait()
close(w)
}()
select {
case <-w:
poolLogger.Infoln("Processes complete")
case <-time.After(t):
poolLogger.Warnf("Timeout")
}
self.flushC = make(chan bool)
}
// 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
// AddPeer can only be used once, RemovePeer needs to be called when the peer disconnects
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()
defer self.peersLock.Unlock()
peer, ok := self.peers[peerId]
if ok {
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.currentBlockHash = currentBlockHash
peer.currentBlock = nil
peer.parentHash = nil
peer.headSection = nil
peer.lock.Unlock()
}
} else {
peer = &peerInfo{
td: td,
currentBlockHash: currentBlockHash,
id: peerId, //peer.Identity().Pubkey()
requestBlockHashes: requestBlockHashes,
requestBlocks: requestBlocks,
peerError: peerError,
sections: make(map[string]*section),
currentBlockC: make(chan *types.Block),
headSectionC: make(chan *section),
}
self.peers[peerId] = peer
poolLogger.Debugf("add new peer %v with td %v and current block %x", peerId, td, currentBlockHash[:4])
}
// check peer current head
if self.hasBlock(currentBlockHash) {
// peer not ahead
return false
}
if self.peer == peer {
// new block update
// peer is already active best peer, request hashes
poolLogger.Debugf("[%s] already the best peer. Request new head section info from %s", peerId, name(currentBlockHash))
peer.headSectionC <- nil
best = true
} else {
currentTD := ethutil.Big0
if self.peer != nil {
currentTD = self.peer.td
}
if td.Cmp(currentTD) > 0 {
poolLogger.Debugf("peer %v promoted best peer", peerId)
self.switchPeer(self.peer, peer)
self.peer = peer
best = true
}
}
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
func (self *BlockPool) RemovePeer(peerId string) {
self.peersLock.Lock()
defer self.peersLock.Unlock()
peer, ok := self.peers[peerId]
if !ok {
return
}
delete(self.peers, peerId)
poolLogger.Debugf("remove peer %v", peerId)
// if current best peer is removed, need find a better one
if self.peer == peer {
var newPeer *peerInfo
max := ethutil.Big0
// peer with the highest self-acclaimed TD is chosen
for _, info := range self.peers {
if info.td.Cmp(max) > 0 {
max = info.td
newPeer = info
}
}
if newPeer != nil {
poolLogger.Debugf("peer %v with td %v promoted to best peer", newPeer.id, newPeer.td)
} else {
poolLogger.Warnln("no peers")
}
self.peer = newPeer
self.switchPeer(peer, newPeer)
}
}
// Entry point for eth protocol to add block hashes received via BlockHashesMsg
// only hashes from the best peer is handled
// this method is always responsible to initiate further hash requests until
// a known parent is reached unless cancelled by a peerChange event
// this process also launches all request processes on each chain section
// this function needs to run asynchronously for one peer since the message is discarded???
func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) {
// register with peer manager loop
peer, best := self.getPeer(peerId)
if !best {
return
}
// peer is still the best
var size, n int
var hash []byte
var ok, headSection bool
var sec, child, parent *section
var entry *poolEntry
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:
// iterate using next (rlp stream lazy decoder) feeding hashesC
for ; ok; hash, ok = next() {
n++
select {
case <-self.quit:
return
case <-quitC:
// if the peer is demoted, no more hashes taken
bestPeer = nil
break LOOP
default:
}
if self.hasBlock(hash) {
// check if known block connecting the downloaded chain to our blockchain
poolLogger.DebugDetailf("[%s] known block", name(hash))
// mark child as absolute pool root with parent known to blockchain
if sec != nil {
self.connectToBlockChain(sec)
} else {
if child != nil {
self.connectToBlockChain(child)
}
}
break LOOP
}
// look up node in pool
entry = self.get(hash)
if entry != nil {
// reached a known chain in the pool
if entry.node == entry.section.bottom && n == 1 {
// 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
continue LOOP
}
poolLogger.DebugDetailf("[%s] reached blockpool chain", name(hash))
parent = entry.section
break LOOP
}
// if node for block hash does not exist, create it and index in the pool
node := &poolNode{
hash: hash,
peer: peerId,
}
if size == 0 {
sec = newSection()
}
nodes = append(nodes, node)
size++
} //for
self.chainLock.Lock()
poolLogger.DebugDetailf("added %v hashes sent by %s", n, peerId)
if parent != nil && entry != nil && entry.node != parent.top {
poolLogger.DebugDetailf("[%s] split section at fork", sectionName(parent))
parent.controlC <- nil
waiter := make(chan bool)
parent.forkC <- waiter
chain := parent.nodes
parent.nodes = chain[entry.index:]
parent.top = parent.nodes[0]
orphan := newSection()
self.link(orphan, parent.child)
self.processSection(orphan, chain[0:entry.index])
orphan.controlC <- nil
close(waiter)
}
if size > 0 {
self.processSection(sec, nodes)
poolLogger.DebugDetailf("[%s]->[%s](%v)->[%s] new chain section", sectionName(parent), sectionName(sec), size, sectionName(child))
self.link(parent, sec)
self.link(sec, child)
} else {
poolLogger.DebugDetailf("[%s]->[%s] connecting known sections", sectionName(parent), sectionName(child))
self.link(parent, child)
}
self.chainLock.Unlock()
if parent != nil && bestPeer != nil {
self.activateChain(parent, peer)
poolLogger.Debugf("[%s] activate parent section [%s]", name(parent.top.hash), sectionName(parent))
}
if sec != nil {
peer.addSection(sec.top.hash, sec)
// request next section here once, only repeat if bottom block arrives,
// 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
}
}
func name(hash []byte) (name string) {
if hash == nil {
name = ""
} else {
name = fmt.Sprintf("%x", hash[:4])
}
return
}
func sectionName(section *section) (name string) {
if section == nil {
name = ""
} else {
name = fmt.Sprintf("%x-%x", section.bottom.hash[:4], section.top.hash[:4])
}
return
}
// AddBlock is the entry point for the eth protocol when blockmsg is received upon requests
// It has a strict interpretation of the protocol in that if the block received has not been requested, it results in an error (which can be ignored)
// block is checked for PoW
// only the first PoW-valid block for a hash is considered legit
func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
hash := block.Hash()
self.peersLock.Lock()
peer := self.peer
self.peersLock.Unlock()
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 {
poolLogger.Warnf("unrequested block [%s] by peer %s", name(hash), peerId)
self.peerError(peerId, ErrUnrequestedBlock, "%x", hash)
}
}
if entry == nil {
return
}
node := entry.node
node.lock.Lock()
defer node.lock.Unlock()
// check if block already present
if node.block != nil {
poolLogger.DebugDetailf("block [%s] already sent by %s", name(hash), node.blockBy)
return
}
if self.hasBlock(hash) {
poolLogger.DebugDetailf("block [%s] already known", name(hash))
} else {
// validate block for PoW
if !self.verifyPoW(block) {
poolLogger.Warnf("invalid pow on block [%s %v] by peer %s", name(hash), block.Number(), peerId)
self.peerError(peerId, ErrInvalidPoW, "%x", hash)
return
}
}
poolLogger.DebugDetailf("added block [%s] sent by peer %s", name(hash), peerId)
node.block = block
node.blockBy = peerId
}
func (self *BlockPool) connectToBlockChain(section *section) {
select {
case <-section.offC:
self.addSectionToBlockChain(section)
case <-section.blockChainC:
default:
close(section.blockChainC)
}
}
func (self *BlockPool) addSectionToBlockChain(section *section) (rest int, err error) {
var blocks types.Blocks
var node *poolNode
var keys []string
rest = len(section.nodes)
for rest > 0 {
rest--
node = section.nodes[rest]
node.lock.RLock()
block := node.block
node.lock.RUnlock()
if block == nil {
break
}
keys = append(keys, string(node.hash))
blocks = append(blocks, block)
}
self.lock.Lock()
for _, key := range keys {
delete(self.pool, key)
}
self.lock.Unlock()
poolLogger.Infof("insert %v blocks into blockchain", len(blocks))
err = self.insertChain(blocks)
if err != nil {
// TODO: not clear which peer we need to address
// peerError should dispatch to peer if still connected and disconnect
self.peerError(node.blockBy, ErrInvalidBlock, "%v", err)
poolLogger.Warnf("invalid block %x", node.hash)
poolLogger.Warnf("penalise peers %v (hash), %v (block)", node.peer, node.blockBy)
// penalise peer in node.blockBy
// self.disconnect()
}
return
}
func (self *BlockPool) activateChain(section *section, peer *peerInfo) {
poolLogger.DebugDetailf("[%s] activate known chain for peer %s", sectionName(section), peer.id)
i := 0
LOOP:
for section != nil {
// register this section with the peer and quit if registered
poolLogger.DebugDetailf("[%s] register section with peer %s", sectionName(section), peer.id)
if peer.addSection(section.top.hash, section) == section {
return
}
poolLogger.DebugDetailf("[%s] activate section process", sectionName(section))
select {
case section.controlC <- peer:
case <-section.offC:
}
i++
section = self.getParent(section)
select {
case <-peer.quitC:
break LOOP
case <-self.quit:
break LOOP
default:
}
}
}
// main worker thread on each section in the poolchain
// - kills the section if there are blocks missing after an absolute time
// - kills the section if there are maxIdleRounds of idle rounds of block requests with no response
// - periodically polls the chain section for missing blocks which are then requested from peers
// - registers the process controller on the peer so that if the peer is promoted as best peer the second time (after a disconnect of a better one), all active processes are switched back on unless they expire and killed ()
// - 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 exits, signals to
func (self *BlockPool) processSection(sec *section, nodes []*poolNode) {
for i, node := range nodes {
entry := &poolEntry{node: node, section: sec, index: i}
self.set(node.hash, entry)
}
sec.bottom = nodes[len(nodes)-1]
sec.top = nodes[0]
sec.nodes = nodes
poolLogger.DebugDetailf("[%s] setup section process", sectionName(sec))
self.wg.Add(1)
go func() {
// absolute time after which sub-chain is killed if not complete (some blocks are missing)
suicideTimer := time.After(blocksTimeout * time.Second)
var peer, newPeer *peerInfo
var blocksRequestTimer, blockHashesRequestTimer <-chan time.Time
var blocksRequestTime, blockHashesRequestTime bool
var blocksRequests, blockHashesRequests int
var blocksRequestsComplete, blockHashesRequestsComplete bool
// node channels for the section
var missingC, processC, offC chan *poolNode
// container for missing block hashes
var hashes [][]byte
var i, missing, lastMissing, depth int
var idle int
var init, done, same, ready bool
var insertChain bool
var quitC chan bool
var blockChainC = sec.blockChainC
var parentHash []byte
LOOP:
for {
if insertChain {
insertChain = false
rest, err := self.addSectionToBlockChain(sec)
if err != nil {
close(sec.suicideC)
continue LOOP
}
if rest == 0 {
blocksRequestsComplete = true
child := self.getChild(sec)
if child != nil {
self.connectToBlockChain(child)
}
}
}
if blockHashesRequestsComplete && blocksRequestsComplete {
// not waiting for hashes any more
poolLogger.Debugf("[%s] section complete %v blocks retrieved (%v attempts), hash requests complete on root (%v attempts)", sectionName(sec), depth, blocksRequests, blockHashesRequests)
break LOOP
} // otherwise suicide if no hashes coming
if done {
// went through all blocks in section
if missing == 0 {
// no missing blocks
poolLogger.DebugDetailf("[%s] got all blocks. process complete (%v total blocksRequests): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
blocksRequestsComplete = true
blocksRequestTimer = nil
blocksRequestTime = false
} else {
poolLogger.DebugDetailf("[%s] section checked: missing %v/%v/%v", sectionName(sec), missing, lastMissing, depth)
// some missing blocks
blocksRequests++
if len(hashes) > 0 {
// send block requests to peers
self.requestBlocks(blocksRequests, hashes)
hashes = nil
}
if missing == lastMissing {
// idle round
if same {
// more than once
idle++
// too many idle rounds
if idle >= blocksRequestMaxIdleRounds {
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(sec.suicideC)
}
} else {
idle = 0
}
same = true
} else {
same = false
}
}
lastMissing = missing
ready = true
done = false
// save a new processC (blocks still missing)
offC = missingC
missingC = processC
// put processC offline
processC = nil
}
//
if ready && blocksRequestTime && !blocksRequestsComplete {
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)
blocksRequestTime = false
processC = offC
}
if blockHashesRequestTime {
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
poolLogger.DebugDetailf("[%s] parent found, hash requests deactivated (after %v total attempts)\n", sectionName(sec), blockHashesRequests)
blockHashesRequestTimer = nil
blockHashesRequestsComplete = true
} else {
blockHashesRequests++
poolLogger.Debugf("[%s] hash request on root (%v total attempts)\n", sectionName(sec), blockHashesRequests)
peer.requestBlockHashes(sec.bottom.hash)
blockHashesRequestTimer = time.After(blockHashesRequestInterval * time.Millisecond)
}
blockHashesRequestTime = false
}
select {
case <-self.quit:
break LOOP
case <-quitC:
// peer quit or demoted, put section in idle mode
quitC = nil
go func() {
sec.controlC <- nil
}()
case <-self.purgeC:
suicideTimer = time.After(0)
case <-suicideTimer:
close(sec.suicideC)
poolLogger.Debugf("[%s] timeout. (%v total attempts): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
case <-sec.suicideC:
poolLogger.Debugf("[%s] suicide", sectionName(sec))
// first delink from child and parent under chainlock
self.chainLock.Lock()
self.link(nil, sec)
self.link(sec, nil)
self.chainLock.Unlock()
// delete node entries from pool index under pool lock
self.lock.Lock()
for _, node := range sec.nodes {
delete(self.pool, string(node.hash))
}
self.lock.Unlock()
break LOOP
case <-blocksRequestTimer:
poolLogger.DebugDetailf("[%s] block request time", sectionName(sec))
blocksRequestTime = true
case <-blockHashesRequestTimer:
poolLogger.DebugDetailf("[%s] hash request time", sectionName(sec))
blockHashesRequestTime = true
case newPeer = <-sec.controlC:
// active -> idle
if peer != nil && newPeer == nil {
self.procWg.Done()
if init {
poolLogger.Debugf("[%s] idle mode (%v total attempts): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
}
blocksRequestTime = false
blocksRequestTimer = nil
blockHashesRequestTime = false
blockHashesRequestTimer = nil
if processC != nil {
offC = processC
processC = nil
}
}
// idle -> active
if peer == nil && newPeer != nil {
self.procWg.Add(1)
poolLogger.Debugf("[%s] active mode", sectionName(sec))
if !blocksRequestsComplete {
blocksRequestTime = true
}
if !blockHashesRequestsComplete && parentHash != nil {
blockHashesRequestTime = true
}
if !init {
processC = make(chan *poolNode, blockHashesBatchSize)
missingC = make(chan *poolNode, blockHashesBatchSize)
i = 0
missing = 0
self.wg.Add(1)
self.procWg.Add(1)
depth = len(sec.nodes)
lastMissing = depth
// if not run at least once fully, launch iterator
go func() {
var node *poolNode
IT:
for _, node = range sec.nodes {
select {
case processC <- node:
case <-self.quit:
break IT
}
}
close(processC)
self.wg.Done()
self.procWg.Done()
}()
} else {
poolLogger.Debugf("[%s] restore earlier state", sectionName(sec))
processC = offC
}
}
// reset quitC to current best peer
if newPeer != nil {
quitC = newPeer.quitC
}
peer = newPeer
case waiter := <-sec.forkC:
// this case just blocks the process until section is split at the fork
<-waiter
init = false
done = false
ready = false
case node, ok := <-processC:
if !ok && !init {
// channel closed, first iteration finished
init = true
done = true
processC = make(chan *poolNode, missing)
poolLogger.DebugDetailf("[%s] section initalised: missing %v/%v/%v", sectionName(sec), missing, lastMissing, depth)
continue LOOP
}
if ready {
i = 0
missing = 0
ready = false
}
i++
// if node has no block
node.lock.RLock()
block := node.block
node.lock.RUnlock()
if block == nil {
missing++
hashes = append(hashes, node.hash)
if len(hashes) == blockBatchSize {
poolLogger.Debugf("[%s] request %v missing blocks", sectionName(sec), len(hashes))
self.requestBlocks(blocksRequests, hashes)
hashes = nil
}
missingC <- node
} else {
if i == lastMissing {
if blockChainC == nil {
insertChain = true
} else {
if parentHash == nil {
parentHash = block.ParentHash()
poolLogger.Debugf("[%s] found root block [%s]", sectionName(sec), name(parentHash))
blockHashesRequestTime = true
}
}
}
}
if i == lastMissing && init {
done = true
}
case <-blockChainC:
// closed blockChain channel indicates that the blockpool is reached
// connected to the blockchain, insert the longest chain of blocks
poolLogger.Debugf("[%s] reached blockchain", sectionName(sec))
blockChainC = nil
// switch off hash requests in case they were on
blockHashesRequestTime = false
blockHashesRequestTimer = nil
blockHashesRequestsComplete = true
// section root has block
if len(sec.nodes) > 0 && sec.nodes[len(sec.nodes)-1].block != nil {
insertChain = true
}
continue LOOP
} // select
} // for
close(sec.offC)
self.wg.Done()
if peer != nil {
self.procWg.Done()
}
}()
return
}
func (self *BlockPool) peerError(peerId string, code int, format string, params ...interface{}) {
self.peersLock.RLock()
defer self.peersLock.RUnlock()
peer, ok := self.peers[peerId]
if ok {
peer.peerError(code, format, params...)
}
}
func (self *BlockPool) requestBlocks(attempts int, hashes [][]byte) {
self.wg.Add(1)
self.procWg.Add(1)
go func() {
// distribute block request among known peers
self.peersLock.Lock()
defer self.peersLock.Unlock()
peerCount := len(self.peers)
// on first attempt use the best peer
if attempts == 0 {
poolLogger.Debugf("request %v missing blocks from best peer %s", len(hashes), self.peer.id)
self.peer.requestBlocks(hashes)
return
}
repetitions := int(math.Min(float64(peerCount), float64(blocksRequestRepetition)))
i := 0
indexes := rand.Perm(peerCount)[0:repetitions]
sort.Ints(indexes)
poolLogger.Debugf("request %v missing blocks from %v/%v peers: chosen %v", len(hashes), repetitions, peerCount, indexes)
for _, peer := range self.peers {
if i == indexes[0] {
poolLogger.Debugf("request %v missing blocks [%x/%x] from peer %s", len(hashes), hashes[0][:4], hashes[len(hashes)-1][:4], peer.id)
peer.requestBlocks(hashes)
indexes = indexes[1:]
if len(indexes) == 0 {
break
}
}
i++
}
self.wg.Done()
self.procWg.Done()
}()
}
func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) {
self.peersLock.RLock()
defer self.peersLock.RUnlock()
if self.peer != nil && self.peer.id == peerId {
return self.peer, true
}
info, ok := self.peers[peerId]
if !ok {
return nil, false
}
return info, false
}
func (self *peerInfo) addSection(hash []byte, section *section) (found *section) {
self.lock.Lock()
defer self.lock.Unlock()
key := string(hash)
found = self.sections[key]
poolLogger.DebugDetailf("[%s] section process stored for %s", sectionName(section), self.id)
self.sections[key] = section
return
}
func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) {
if newPeer != nil {
newPeer.quitC = make(chan bool)
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 {
// this will block if section process is waiting for peer lock
select {
case <-section.offC:
poolLogger.DebugDetailf("[%s][%x] section process complete - remove", newPeer.id, hash[:4])
delete(newPeer.sections, hash)
case section.controlC <- newPeer:
poolLogger.DebugDetailf("[%s][%x] activates section [%s]", newPeer.id, hash[:4], sectionName(section))
}
}
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 {
poolLogger.DebugDetailf("[%s] quit section processes", oldPeer.id)
close(oldPeer.quitC)
}
}
func (self *BlockPool) getParent(sec *section) *section {
self.chainLock.RLock()
defer self.chainLock.RUnlock()
return sec.parent
}
func (self *BlockPool) getChild(sec *section) *section {
self.chainLock.RLock()
defer self.chainLock.RUnlock()
return sec.child
}
func newSection() (sec *section) {
sec = &section{
controlC: make(chan *peerInfo),
suicideC: make(chan bool),
blockChainC: make(chan bool),
offC: make(chan bool),
forkC: make(chan chan bool),
}
return
}
// link should only be called under chainLock
func (self *BlockPool) link(parent *section, child *section) {
if parent != nil {
exChild := parent.child
parent.child = child
if exChild != nil && exChild != child {
poolLogger.Debugf("[%s] chain fork [%s] -> [%s]", sectionName(parent), sectionName(exChild), sectionName(child))
exChild.parent = nil
}
}
if child != nil {
exParent := child.parent
if exParent != nil && exParent != parent {
poolLogger.Debugf("[%s] chain reverse fork [%s] -> [%s]", sectionName(child), sectionName(exParent), sectionName(parent))
exParent.child = nil
}
child.parent = parent
}
}
func (self *BlockPool) get(hash []byte) (node *poolEntry) {
self.lock.RLock()
defer self.lock.RUnlock()
return self.pool[string(hash)]
}
func (self *BlockPool) set(hash []byte, node *poolEntry) {
self.lock.Lock()
defer self.lock.Unlock()
self.pool[string(hash)] = node
}
eth/block_pool_test.go deleted 100644 → 0
View file @ cdc86965
package eth
import (
"fmt"
"log"
"math/big"
"os"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethutil"
ethlogger "github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/pow"
)
const waitTimeout = 60 // seconds
var logsys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel))
var ini = false
func logInit() {
if !ini {
ethlogger.AddLogSystem(logsys)
ini = true
}
}
// test helpers
func arrayEq(a, b []int) bool {
if len(a) != len(b) {
return false
}
for i := range a {
if a[i] != b[i] {
return false
}
}
return true
}
type intToHash map[int][]byte
type hashToInt map[string]int
// hashPool is a test helper, that allows random hashes to be referred to by integers
type testHashPool struct {
intToHash
hashToInt
lock sync.Mutex
}
func newHash(i int) []byte {
return crypto.Sha3([]byte(string(i)))
}
func (self *testHashPool) indexesToHashes(indexes []int) (hashes [][]byte) {
self.lock.Lock()
defer self.lock.Unlock()
for _, i := range indexes {
hash, found := self.intToHash[i]
if !found {
hash = newHash(i)
self.intToHash[i] = hash
self.hashToInt[string(hash)] = i
}
hashes = append(hashes, hash)
}
return
}
func (self *testHashPool) hashesToIndexes(hashes [][]byte) (indexes []int) {
self.lock.Lock()
defer self.lock.Unlock()
for _, hash := range hashes {
i, found := self.hashToInt[string(hash)]
if !found {
i = -1
}
indexes = append(indexes, i)
}
return
}
// test blockChain is an integer trie
type blockChain map[int][]int
// blockPoolTester provides the interface between tests and a blockPool
//
// refBlockChain is used to guide which blocks will be accepted as valid
// blockChain gives the current state of the blockchain and
// accumulates inserts so that we can check the resulting chain
type blockPoolTester struct {
hashPool *testHashPool
lock sync.RWMutex
refBlockChain blockChain
blockChain blockChain
blockPool *BlockPool
t *testing.T
}
func newTestBlockPool(t *testing.T) (hashPool *testHashPool, blockPool *BlockPool, b *blockPoolTester) {
hashPool = &testHashPool{intToHash: make(intToHash), hashToInt: make(hashToInt)}
b = &blockPoolTester{
t: t,
hashPool: hashPool,
blockChain: make(blockChain),
refBlockChain: make(blockChain),
}
b.blockPool = NewBlockPool(b.hasBlock, b.insertChain, b.verifyPoW)
blockPool = b.blockPool
return
}
func (self *blockPoolTester) Errorf(format string, params ...interface{}) {
fmt.Printf(format+"\n", params...)
self.t.Errorf(format, params...)
}
// blockPoolTester implements the 3 callbacks needed by the blockPool:
// hasBlock, insetChain, verifyPoW
func (self *blockPoolTester) hasBlock(block []byte) (ok bool) {
self.lock.RLock()
defer self.lock.RUnlock()
indexes := self.hashPool.hashesToIndexes([][]byte{block})
i := indexes[0]
_, ok = self.blockChain[i]
fmt.Printf("has block %v (%x...): %v\n", i, block[0:4], ok)
return
}
func (self *blockPoolTester) insertChain(blocks types.Blocks) error {
self.lock.RLock()
defer self.lock.RUnlock()
var parent, child int
var children, refChildren []int
var ok bool
for _, block := range blocks {
child = self.hashPool.hashesToIndexes([][]byte{block.Hash()})[0]
_, ok = self.blockChain[child]
if ok {
fmt.Printf("block %v already in blockchain\n", child)
continue // already in chain
}
parent = self.hashPool.hashesToIndexes([][]byte{block.ParentHeaderHash})[0]
children, ok = self.blockChain[parent]
if !ok {
return fmt.Errorf("parent %v not in blockchain ", parent)
}
ok = false
var found bool
refChildren, found = self.refBlockChain[parent]
if found {
for _, c := range refChildren {
if c == child {
ok = true
}
}
if !ok {
return fmt.Errorf("invalid block %v", child)
}
} else {
ok = true
}
if ok {
// accept any blocks if parent not in refBlockChain
fmt.Errorf("blockchain insert %v -> %v\n", parent, child)
self.blockChain[parent] = append(children, child)
self.blockChain[child] = nil
}
}
return nil
}
func (self *blockPoolTester) verifyPoW(pblock pow.Block) bool {
return true
}
// test helper that compares the resulting blockChain to the desired blockChain
func (self *blockPoolTester) checkBlockChain(blockChain map[int][]int) {
for k, v := range self.blockChain {
fmt.Printf("got: %v -> %v\n", k, v)
}
for k, v := range blockChain {
fmt.Printf("expected: %v -> %v\n", k, v)
}
if len(blockChain) != len(self.blockChain) {
self.Errorf("blockchain incorrect (zlength differ)")
}
for k, v := range blockChain {
vv, ok := self.blockChain[k]
if !ok || !arrayEq(v, vv) {
self.Errorf("blockchain incorrect on %v -> %v (!= %v)", k, vv, v)
}
}
}
//
// peerTester provides the peer callbacks for the blockPool
// it registers actual callbacks so that result can be compared to desired behaviour
// provides helper functions to mock the protocol calls to the blockPool
type peerTester struct {
blockHashesRequests []int
blocksRequests [][]int
blocksRequestsMap map[int]bool
peerErrors []int
blockPool *BlockPool
hashPool *testHashPool
lock sync.RWMutex
id string
td int
currentBlock int
t *testing.T
}
// peerTester constructor takes hashPool and blockPool from the blockPoolTester
func (self *blockPoolTester) newPeer(id string, td int, cb int) *peerTester {
return &peerTester{
id: id,
td: td,
currentBlock: cb,
hashPool: self.hashPool,
blockPool: self.blockPool,
t: self.t,
blocksRequestsMap: make(map[int]bool),
}
}
func (self *peerTester) Errorf(format string, params ...interface{}) {
fmt.Printf(format+"\n", params...)
self.t.Errorf(format, params...)
}
// helper to compare actual and expected block requests
func (self *peerTester) checkBlocksRequests(blocksRequests ...[]int) {
if len(blocksRequests) > len(self.blocksRequests) {
self.Errorf("blocks requests incorrect (length differ)\ngot %v\nexpected %v", self.blocksRequests, blocksRequests)
} else {
for i, rr := range blocksRequests {
r := self.blocksRequests[i]
if !arrayEq(r, rr) {
self.Errorf("blocks requests incorrect\ngot %v\nexpected %v", self.blocksRequests, blocksRequests)
}
}
}
}
// helper to compare actual and expected block hash requests
func (self *peerTester) checkBlockHashesRequests(blocksHashesRequests ...int) {
rr := blocksHashesRequests
self.lock.RLock()
r := self.blockHashesRequests
self.lock.RUnlock()
if len(r) != len(rr) {
self.Errorf("block hashes requests incorrect (length differ)\ngot %v\nexpected %v", r, rr)
} else {
if !arrayEq(r, rr) {
self.Errorf("block hashes requests incorrect\ngot %v\nexpected %v", r, rr)
}
}
}
// waiter function used by peer.AddBlocks
// blocking until requests appear
// since block requests are sent to any random peers
// block request map is shared between peers
// times out after a period
func (self *peerTester) waitBlocksRequests(blocksRequest ...int) {
timeout := time.After(waitTimeout * time.Second)
rr := blocksRequest
for {
self.lock.RLock()
r := self.blocksRequestsMap
fmt.Printf("[%s] blocks request check %v (%v)\n", self.id, rr, r)
i := 0
for i = 0; i < len(rr); i++ {
_, ok := r[rr[i]]
if !ok {
break
}
}
self.lock.RUnlock()
if i == len(rr) {
return
}
time.Sleep(100 * time.Millisecond)
select {
case <-timeout:
default:
}
}
}
// waiter function used by peer.AddBlockHashes
// blocking until requests appear
// times out after a period
func (self *peerTester) waitBlockHashesRequests(blocksHashesRequest int) {
timeout := time.After(waitTimeout * time.Second)
rr := blocksHashesRequest
for i := 0; ; {
self.lock.RLock()
r := self.blockHashesRequests
self.lock.RUnlock()
fmt.Printf("[%s] block hash request check %v (%v)\n", self.id, rr, r)
for ; i < len(r); i++ {
if rr == r[i] {
return
}
}
time.Sleep(100 * time.Millisecond)
select {
case <-timeout:
default:
}
}
}
// mocks a simple blockchain 0 (genesis) ... n (head)
func (self *blockPoolTester) initRefBlockChain(n int) {
for i := 0; i < n; i++ {
self.refBlockChain[i] = []int{i + 1}
}
}
// peerTester functions that mimic protocol calls to the blockpool
// registers the peer with the blockPool
func (self *peerTester) AddPeer() bool {
hash := self.hashPool.indexesToHashes([]int{self.currentBlock})[0]
return self.blockPool.AddPeer(big.NewInt(int64(self.td)), hash, self.id, self.requestBlockHashes, self.requestBlocks, self.peerError)
}
// peer sends blockhashes if and when gets a request
func (self *peerTester) AddBlockHashes(indexes ...int) {
fmt.Printf("ready to add block hashes %v\n", indexes)
self.waitBlockHashesRequests(indexes[0])
fmt.Printf("adding block hashes %v\n", indexes)
hashes := self.hashPool.indexesToHashes(indexes)
i := 1
next := func() (hash []byte, ok bool) {
if i < len(hashes) {
hash = hashes[i]
ok = true
i++
}
return
}
self.blockPool.AddBlockHashes(next, self.id)
}
// peer sends blocks if and when there is a request
// (in the shared request store, not necessarily to a person)
func (self *peerTester) AddBlocks(indexes ...int) {
hashes := self.hashPool.indexesToHashes(indexes)
fmt.Printf("ready to add blocks %v\n", indexes[1:])
self.waitBlocksRequests(indexes[1:]...)
fmt.Printf("adding blocks %v \n", indexes[1:])
for i := 1; i < len(hashes); i++ {
fmt.Printf("adding block %v %x\n", indexes[i], hashes[i][:4])
self.blockPool.AddBlock(&types.Block{HeaderHash: ethutil.Bytes(hashes[i]), ParentHeaderHash: ethutil.Bytes(hashes[i-1])}, self.id)
}
}
// peer callbacks
// -1 is special: not found (a hash never seen)
// records block hashes requests by the blockPool
func (self *peerTester) requestBlockHashes(hash []byte) error {
indexes := self.hashPool.hashesToIndexes([][]byte{hash})
fmt.Printf("[%s] blocks hash request %v %x\n", self.id, indexes[0], hash[:4])
self.lock.Lock()
defer self.lock.Unlock()
self.blockHashesRequests = append(self.blockHashesRequests, indexes[0])
return nil
}
// records block requests by the blockPool
func (self *peerTester) requestBlocks(hashes [][]byte) error {
indexes := self.hashPool.hashesToIndexes(hashes)
fmt.Printf("blocks request %v %x...\n", indexes, hashes[0][:4])
self.lock.Lock()
defer self.lock.Unlock()
self.blocksRequests = append(self.blocksRequests, indexes)
for _, i := range indexes {
self.blocksRequestsMap[i] = true
}
return nil
}
// records the error codes of all the peerErrors found the blockPool
func (self *peerTester) peerError(code int, format string, params ...interface{}) {
self.peerErrors = append(self.peerErrors, code)
}
// the actual tests
func TestAddPeer(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
peer0 := blockPoolTester.newPeer("peer0", 1, 0)
peer1 := blockPoolTester.newPeer("peer1", 2, 1)
peer2 := blockPoolTester.newPeer("peer2", 3, 2)
var peer *peerInfo
blockPool.Start()
// pool
best := peer0.AddPeer()
if !best {
t.Errorf("peer0 (TD=1) not accepted as best")
}
if blockPool.peer.id != "peer0" {
t.Errorf("peer0 (TD=1) not set as best")
}
// peer0.checkBlockHashesRequests(0)
best = peer2.AddPeer()
if !best {
t.Errorf("peer2 (TD=3) not accepted as best")
}
if blockPool.peer.id != "peer2" {
t.Errorf("peer2 (TD=3) not set as best")
}
peer2.waitBlocksRequests(2)
best = peer1.AddPeer()
if best {
t.Errorf("peer1 (TD=2) accepted as best")
}
if blockPool.peer.id != "peer2" {
t.Errorf("peer2 (TD=3) not set any more as best")
}
if blockPool.peer.td.Cmp(big.NewInt(int64(3))) != 0 {
t.Errorf("peer1 TD not set")
}
peer2.td = 4
peer2.currentBlock = 3
best = peer2.AddPeer()
if !best {
t.Errorf("peer2 (TD=4) not accepted as best")
}
if blockPool.peer.id != "peer2" {
t.Errorf("peer2 (TD=4) not set as best")
}
if blockPool.peer.td.Cmp(big.NewInt(int64(4))) != 0 {
t.Errorf("peer2 TD not updated")
}
peer2.waitBlocksRequests(3)
peer1.td = 3
peer1.currentBlock = 2
best = peer1.AddPeer()
if best {
t.Errorf("peer1 (TD=3) should not be set as best")
}
if blockPool.peer.id == "peer1" {
t.Errorf("peer1 (TD=3) should not be set as best")
}
peer, best = blockPool.getPeer("peer1")
if peer.td.Cmp(big.NewInt(int64(3))) != 0 {
t.Errorf("peer1 TD should be updated")
}
blockPool.RemovePeer("peer2")
peer, best = blockPool.getPeer("peer2")
if peer != nil {
t.Errorf("peer2 not removed")
}
if blockPool.peer.id != "peer1" {
t.Errorf("existing peer1 (TD=3) should be set as best peer")
}
peer1.waitBlocksRequests(2)
blockPool.RemovePeer("peer1")
peer, best = blockPool.getPeer("peer1")
if peer != nil {
t.Errorf("peer1 not removed")
}
if blockPool.peer.id != "peer0" {
t.Errorf("existing peer0 (TD=1) should be set as best peer")
}
peer0.waitBlocksRequests(0)
blockPool.RemovePeer("peer0")
peer, best = blockPool.getPeer("peer0")
if peer != nil {
t.Errorf("peer1 not removed")
}
// adding back earlier peer ok
peer0.currentBlock = 3
best = peer0.AddPeer()
if !best {
t.Errorf("peer0 (TD=1) should be set as best")
}
if blockPool.peer.id != "peer0" {
t.Errorf("peer0 (TD=1) should be set as best")
}
peer0.waitBlocksRequests(3)
blockPool.Stop()
}
func TestPeerWithKnownBlock(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.refBlockChain[0] = nil
blockPoolTester.blockChain[0] = nil
blockPool.Start()
peer0 := blockPoolTester.newPeer("0", 1, 0)
peer0.AddPeer()
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
// no request on known block
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) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(2)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 2)
peer1.AddPeer()
peer1.AddBlocks(1, 2)
go peer1.AddBlockHashes(2, 1, 0)
peer1.AddBlocks(0, 1)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[2] = []int{}
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) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(2)
blockPoolTester.refBlockChain[2] = []int{}
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 3)
peer1.AddPeer()
go peer1.AddBlocks(2, 3)
go peer1.AddBlockHashes(3, 2, 1, 0)
peer1.AddBlocks(0, 1, 2)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[2] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
if len(peer1.peerErrors) == 1 {
if peer1.peerErrors[0] != ErrInvalidBlock {
t.Errorf("wrong error, got %v, expected %v", peer1.peerErrors[0], ErrInvalidBlock)
}
} else {
t.Errorf("expected invalid block error, got nothing %v", peer1.peerErrors)
}
}
func TestVerifyPoW(t *testing.T) {
t.Skip("***FIX*** This test is broken")
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(3)
first := false
blockPoolTester.blockPool.verifyPoW = func(b pow.Block) bool {
bb, _ := b.(*types.Block)
indexes := blockPoolTester.hashPool.hashesToIndexes([][]byte{bb.Hash()})
if indexes[0] == 2 && !first {
first = true
return false
} else {
return true
}
}
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 3)
peer1.AddPeer()
go peer1.AddBlocks(2, 3)
go peer1.AddBlockHashes(3, 2, 1, 0)
peer1.AddBlocks(0, 1, 2)
// blockPool.Wait(waitTimeout * time.Second)
time.Sleep(1 * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[1] = []int{}
delete(blockPoolTester.refBlockChain, 2)
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
if len(peer1.peerErrors) == 1 {
if peer1.peerErrors[0] != ErrInvalidPoW {
t.Errorf("wrong error, got %v, expected %v", peer1.peerErrors[0], ErrInvalidPoW)
}
} else {
t.Errorf("expected invalid pow error, got nothing")
}
}
func TestMultiSectionChain(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(5)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 5)
peer1.AddPeer()
go peer1.AddBlocks(4, 5)
go peer1.AddBlockHashes(5, 4, 3)
go peer1.AddBlocks(2, 3, 4)
go peer1.AddBlockHashes(3, 2, 1, 0)
peer1.AddBlocks(0, 1, 2)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[5] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestNewBlocksOnPartialChain(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(7)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 5)
peer1.AddPeer()
go peer1.AddBlocks(4, 5) // partially complete section
go peer1.AddBlockHashes(5, 4, 3)
peer1.AddBlocks(3, 4) // partially complete section
// peer1 found new blocks
peer1.td = 2
peer1.currentBlock = 7
peer1.AddPeer()
go peer1.AddBlocks(6, 7)
go peer1.AddBlockHashes(7, 6, 5)
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
peer1.AddBlocks(0, 1, 2)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[7] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestPeerSwitchUp(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(7)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 6)
peer2 := blockPoolTester.newPeer("peer2", 2, 7)
peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer1.AddPeer()
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
peer2.AddPeer() // peer2 is promoted as best peer, peer1 is demoted
go peer2.AddBlocks(6, 7)
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 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
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[7] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestPeerSwitchDown(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(4, 5) //
blockPool.RemovePeer("peer2") // peer2 disconnects
peer1.AddPeer() // inferior peer1 is promoted as best peer
go peer1.AddBlockHashes(4, 3, 2, 1, 0) //
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)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[6] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestPeerSwitchBack(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(8)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 2, 11)
peer2 := blockPoolTester.newPeer("peer2", 1, 8)
peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer2.AddPeer()
go peer2.AddBlocks(7, 8)
go peer2.AddBlockHashes(8, 7, 6)
go peer2.AddBlockHashes(6, 5, 4)
peer2.AddBlocks(4, 5) // section partially complete
peer1.AddPeer() // peer1 is promoted as best peer
go peer1.AddBlocks(10, 11) //
peer1.AddBlockHashes(11, 10) // only gives useless results
blockPool.RemovePeer("peer1") // peer1 disconnects
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
peer2.AddBlocks(0, 1, 2, 3)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[8] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestForkSimple(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(9)
blockPoolTester.refBlockChain[3] = []int{4, 7}
delete(blockPoolTester.refBlockChain, 6)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 9)
peer2 := blockPoolTester.newPeer("peer2", 2, 6)
peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer1.AddPeer()
go peer1.AddBlocks(8, 9)
go peer1.AddBlockHashes(9, 8, 7, 3, 2)
peer1.AddBlocks(1, 2, 3, 7, 8)
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.AddBlocks(1, 2, 3, 4, 5)
go peer2.AddBlockHashes(2, 1, 0)
peer2.AddBlocks(0, 1, 2)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[6] = []int{}
blockPoolTester.refBlockChain[3] = []int{4}
delete(blockPoolTester.refBlockChain, 7)
delete(blockPoolTester.refBlockChain, 8)
delete(blockPoolTester.refBlockChain, 9)
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestForkSwitchBackByNewBlocks(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(11)
blockPoolTester.refBlockChain[3] = []int{4, 7}
delete(blockPoolTester.refBlockChain, 6)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 9)
peer2 := blockPoolTester.newPeer("peer2", 2, 6)
peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer1.AddPeer()
peer1.AddBlocks(8, 9) //
go peer1.AddBlockHashes(9, 8, 7, 3, 2) //
peer1.AddBlocks(7, 8) // partial section
peer2.AddPeer() //
peer2.AddBlocks(5, 6) //
go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
peer2.AddBlocks(1, 2, 3, 4, 5) //
// peer1 finds new blocks
peer1.td = 3
peer1.currentBlock = 11
peer1.AddPeer()
go peer1.AddBlocks(10, 11)
go peer1.AddBlockHashes(11, 10, 9)
peer1.AddBlocks(9, 10)
go peer1.AddBlocks(3, 7) // tests that block requests on earlier fork are 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)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[11] = []int{}
blockPoolTester.refBlockChain[3] = []int{7}
delete(blockPoolTester.refBlockChain, 6)
delete(blockPoolTester.refBlockChain, 5)
delete(blockPoolTester.refBlockChain, 4)
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestForkSwitchBackByPeerSwitchBack(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(9)
blockPoolTester.refBlockChain[3] = []int{4, 7}
delete(blockPoolTester.refBlockChain, 6)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 9)
peer2 := blockPoolTester.newPeer("peer2", 2, 6)
peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer1.AddPeer()
go peer1.AddBlocks(8, 9)
go peer1.AddBlockHashes(9, 8, 7, 3, 2)
peer1.AddBlocks(7, 8)
peer2.AddPeer()
go peer2.AddBlocks(5, 6) //
go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
peer2.AddBlocks(2, 3, 4, 5) //
blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer
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(1, 2) //
go peer1.AddBlockHashes(2, 1, 0) //
peer1.AddBlocks(0, 1)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[9] = []int{}
blockPoolTester.refBlockChain[3] = []int{7}
delete(blockPoolTester.refBlockChain, 6)
delete(blockPoolTester.refBlockChain, 5)
delete(blockPoolTester.refBlockChain, 4)
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestForkCompleteSectionSwitchBackByPeerSwitchBack(t *testing.T) {
logInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(9)
blockPoolTester.refBlockChain[3] = []int{4, 7}
delete(blockPoolTester.refBlockChain, 6)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 9)
peer2 := blockPoolTester.newPeer("peer2", 2, 6)
peer2.blocksRequestsMap = peer1.blocksRequestsMap
peer1.AddPeer()
go peer1.AddBlocks(8, 9)
go peer1.AddBlockHashes(9, 8, 7)
peer1.AddBlocks(3, 7, 8) // make sure this section is complete
time.Sleep(1 * time.Second)
go peer1.AddBlockHashes(7, 3, 2) // block 3/7 is section boundary
peer1.AddBlocks(2, 3) // partially complete sections block 2 missing
peer2.AddPeer() //
go peer2.AddBlocks(5, 6) //
go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
peer2.AddBlocks(2, 3, 4, 5) // block 2 still missing.
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
go peer1.AddBlockHashes(2, 1, 0) //
peer1.AddBlocks(0, 1, 2)
blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
blockPoolTester.refBlockChain[9] = []int{}
blockPoolTester.refBlockChain[3] = []int{7}
delete(blockPoolTester.refBlockChain, 6)
delete(blockPoolTester.refBlockChain, 5)
delete(blockPoolTester.refBlockChain, 4)
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
eth/error.go deleted 100644 → 0
View file @ cdc86965
package eth
import (
"fmt"
)
const (
ErrMsgTooLarge = iota
ErrDecode
ErrInvalidMsgCode
ErrProtocolVersionMismatch
ErrNetworkIdMismatch
ErrGenesisBlockMismatch
ErrNoStatusMsg
ErrExtraStatusMsg
ErrInvalidBlock
ErrInvalidPoW
ErrUnrequestedBlock
ErrInsufficientChainInfo
)
var errorToString = map[int]string{
ErrMsgTooLarge: "Message too long",
ErrDecode: "Invalid message",
ErrInvalidMsgCode: "Invalid message code",
ErrProtocolVersionMismatch: "Protocol version mismatch",
ErrNetworkIdMismatch: "NetworkId mismatch",
ErrGenesisBlockMismatch: "Genesis block mismatch",
ErrNoStatusMsg: "No status message",
ErrExtraStatusMsg: "Extra status message",
ErrInvalidBlock: "Invalid block",
ErrInvalidPoW: "Invalid PoW",
ErrUnrequestedBlock: "Unrequested block",
ErrInsufficientChainInfo: "Insufficient chain info",
}
type protocolError struct {
Code int
fatal bool
message string
format string
params []interface{}
// size int
}
func newProtocolError(code int, format string, params ...interface{}) *protocolError {
return &protocolError{Code: code, format: format, params: params}
}
func ProtocolError(code int, format string, params ...interface{}) (err *protocolError) {
err = newProtocolError(code, format, params...)
// report(err)
return
}
func (self protocolError) Error() (message string) {
if len(message) == 0 {
var ok bool
self.message, ok = errorToString[self.Code]
if !ok {
panic("invalid error code")
}
if self.format != "" {
self.message += ": " + fmt.Sprintf(self.format, self.params...)
}
}
return self.message
}
func (self *protocolError) Fatal() bool {
return self.fatal
}
eth/protocol.go
View file @ 16ecda95
......@@ -7,6 +7,7 @@ import (
"math/big"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/errs"
"github.com/ethereum/go-ethereum/ethutil"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rlp"
......@@ -17,6 +18,8 @@ const (
NetworkId = 0
ProtocolLength = uint64(8)
ProtocolMaxMsgSize = 10 * 1024 * 1024
maxHashes = 256
maxBlocks = 64
)
// eth protocol message codes
......@@ -31,6 +34,28 @@ const (
NewBlockMsg
)
const (
ErrMsgTooLarge = iota
ErrDecode
ErrInvalidMsgCode
ErrProtocolVersionMismatch
ErrNetworkIdMismatch
ErrGenesisBlockMismatch
ErrNoStatusMsg
ErrExtraStatusMsg
)
var errorToString = map[int]string{
ErrMsgTooLarge: "Message too long",
ErrDecode: "Invalid message",
ErrInvalidMsgCode: "Invalid message code",
ErrProtocolVersionMismatch: "Protocol version mismatch",
ErrNetworkIdMismatch: "NetworkId mismatch",
ErrGenesisBlockMismatch: "Genesis block mismatch",
ErrNoStatusMsg: "No status message",
ErrExtraStatusMsg: "Extra status message",
}
// ethProtocol represents the ethereum wire protocol
// instance is running on each peer
type ethProtocol struct {
......@@ -40,6 +65,7 @@ type ethProtocol struct {
peer *p2p.Peer
id string
rw p2p.MsgReadWriter
errors *errs.Errors
}
// backend is the interface the ethereum protocol backend should implement
......@@ -58,7 +84,7 @@ type chainManager interface {
type blockPool interface {
AddBlockHashes(next func() ([]byte, bool), peerId string)
AddBlock(block *types.Block, peerId string)
AddPeer(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) (best bool)
AddPeer(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(*errs.Error)) (best bool)
RemovePeer(peerId string)
}
......@@ -68,8 +94,6 @@ type newBlockMsgData struct {
TD *big.Int
}
const maxHashes = 255
type getBlockHashesMsgData struct {
Hash []byte
Amount uint64
......@@ -99,7 +123,11 @@ func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPoo
blockPool: blockPool,
rw: rw,
peer: peer,
id: fmt.Sprintf("%x", id[:8]),
errors: &errs.Errors{
Package: "ETH",
Errors: errorToString,
},
id: fmt.Sprintf("%x", id[:8]),
}
err = self.handleStatus()
if err == nil {
......@@ -145,7 +173,6 @@ func (self *ethProtocol) handle() error {
return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
}
//request.Amount = uint64(math.Min(float64(maxHashes), float64(request.Amount)))
if request.Amount > maxHashes {
request.Amount = maxHashes
}
......@@ -153,7 +180,6 @@ func (self *ethProtocol) handle() error {
return p2p.EncodeMsg(self.rw, BlockHashesMsg, ethutil.ByteSliceToInterface(hashes)...)
case BlockHashesMsg:
// TODO: redo using lazy decode , this way very inefficient on known chains
msgStream := rlp.NewStream(msg.Payload)
var err error
var i int
......@@ -191,7 +217,7 @@ func (self *ethProtocol) handle() error {
if block != nil {
blocks = append(blocks, block)
}
if i == blockHashesBatchSize {
if i == maxBlocks {
break
}
}
......@@ -218,7 +244,7 @@ func (self *ethProtocol) handle() error {
}
hash := request.Block.Hash()
// to simplify backend interface adding a new block
// uses AddPeer followed by AddHashes, AddBlock only if peer is the best peer
// uses AddPeer followed by AddBlock only if peer is the best peer
// (or selected as new best peer)
if self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) {
self.blockPool.AddBlock(request.Block, self.id)
......@@ -277,7 +303,7 @@ func (self *ethProtocol) handleStatus() error {
_, _, genesisBlock := self.chainManager.Status()
if bytes.Compare(status.GenesisBlock, genesisBlock) != 0 {
if !bytes.Equal(status.GenesisBlock, genesisBlock) {
return self.protoError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock)
}
......@@ -297,8 +323,8 @@ func (self *ethProtocol) handleStatus() error {
}
func (self *ethProtocol) requestBlockHashes(from []byte) error {
self.peer.Debugf("fetching hashes (%d) %x...\n", blockHashesBatchSize, from[0:4])
return p2p.EncodeMsg(self.rw, GetBlockHashesMsg, interface{}(from), uint64(blockHashesBatchSize))
self.peer.Debugf("fetching hashes (%d) %x...\n", maxHashes, from[0:4])
return p2p.EncodeMsg(self.rw, GetBlockHashesMsg, interface{}(from), uint64(maxHashes))
}
func (self *ethProtocol) requestBlocks(hashes [][]byte) error {
......@@ -306,26 +332,17 @@ func (self *ethProtocol) requestBlocks(hashes [][]byte) error {
return p2p.EncodeMsg(self.rw, GetBlocksMsg, ethutil.ByteSliceToInterface(hashes)...)
}
func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *protocolError) {
err = ProtocolError(code, format, params...)
if err.Fatal() {
self.peer.Errorln("err %v", err)
// disconnect
} else {
self.peer.Debugf("fyi %v", err)
}
func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *errs.Error) {
err = self.errors.New(code, format, params...)
err.Log(self.peer.Logger)
return
}
func (self *ethProtocol) protoErrorDisconnect(code int, format string, params ...interface{}) {
err := ProtocolError(code, format, params...)
func (self *ethProtocol) protoErrorDisconnect(err *errs.Error) {
err.Log(self.peer.Logger)
if err.Fatal() {
self.peer.Errorln("err %v", err)
// disconnect
} else {
self.peer.Debugf("fyi %v", err)
self.peer.Disconnect(p2p.DiscSubprotocolError)
}
}
func (self *ethProtocol) propagateTxs() {
......
eth/protocol_test.go
View file @ 16ecda95
......@@ -11,13 +11,23 @@ import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/errs"
"github.com/ethereum/go-ethereum/ethutil"
ethlogger "github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
)
var sys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel))
var logsys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel))
var ini = false
func logInit() {
if !ini {
ethlogger.AddLogSystem(logsys)
ini = true
}
}
type testMsgReadWriter struct {
in chan p2p.Msg
......@@ -64,7 +74,7 @@ type testChainManager struct {
type testBlockPool struct {
addBlockHashes func(next func() ([]byte, bool), peerId string)
addBlock func(block *types.Block, peerId string) (err error)
addPeer func(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) (best bool)
addPeer func(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(*errs.Error)) (best bool)
removePeer func(peerId string)
}
......@@ -116,7 +126,7 @@ func (self *testBlockPool) AddBlock(block *types.Block, peerId string) {
}
}
func (self *testBlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) (best bool) {
func (self *testBlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(*errs.Error)) (best bool) {
if self.addPeer != nil {
best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError)
}
......@@ -169,7 +179,7 @@ func (self *ethProtocolTester) checkError(expCode int, delay time.Duration) (err
self.t.Errorf("no error after %v, expected %v", delay, expCode)
return
}
perr, ok := err.(*protocolError)
perr, ok := err.(*errs.Error)
if ok && perr != nil {
if code := perr.Code; code != expCode {
self.t.Errorf("expected protocol error (code %v), got %v (%v)", expCode, code, err)
......
eth/test/README.md deleted 100644 → 0
View file @ cdc86965
= Integration tests for eth protocol and blockpool
This is a simple suite of tests to fire up a local test node with peers to test blockchain synchronisation and download.
The scripts call ethereum (assumed to be compiled in go-ethereum root).
To run a test:
. run.sh 00 02
Without arguments, all tests are run.
Peers are launched with preloaded imported chains. In order to prevent them from synchronizing with each other they are set with `-dial=false` and `-maxpeer 1` options. They log into `/tmp/eth.test/nodes/XX` where XX is the last two digits of their port.
Chains to import can be bootstrapped by letting nodes mine for some time. This is done with
. bootstrap.sh
Only the relative timing and forks matter so they should work if the bootstrap script is rerun.
The reference blockchain of tests are soft links to these import chains and check at the end of a test run.
Connecting to peers and exporting blockchain is scripted with JS files executed by the JSRE, see `tests/XX.sh`.
Each test is set with a timeout. This may vary on different computers so adjust sensibly.
If you kill a test before it completes, do not forget to kill all the background processes, since they will impact the result. Use:
killall ethereum
eth/test/bootstrap.sh deleted 100644 → 0
View file @ cdc86965
#!/bin/bash
# bootstrap chains - used to regenerate tests/chains/*.chain
mkdir -p chains
bash ./mine.sh 00 10
bash ./mine.sh 01 5 00
bash ./mine.sh 02 10 00
bash ./mine.sh 03 5 02
bash ./mine.sh 04 10 02
\ No newline at end of file
eth/test/mine.sh deleted 100644 → 0
View file @ cdc86965
#!/bin/bash
# bash ./mine.sh node_id timeout(sec) [basechain]
ETH=../../ethereum
MINE="$ETH -datadir tmp/nodes/$1 -seed=false -port '' -shh=false -id test$1"
rm -rf tmp/nodes/$1
echo "Creating chain $1..."
if [[ "" != "$3" ]]; then
CHAIN="chains/$3.chain"
CHAINARG="-chain $CHAIN"
$MINE -mine $CHAINARG -loglevel 3 | grep 'importing'
fi
$MINE -mine -loglevel 0 &
PID=$!
sleep $2
kill $PID
$MINE -loglevel 3 <(echo "eth.export(\"chains/$1.chain\")") > /tmp/eth.test/mine.tmp &
PID=$!
sleep 1
kill $PID
cat /tmp/eth.test/mine.tmp | grep 'exporting'
eth/test/run.sh deleted 100644 → 0
View file @ cdc86965
#!/bin/bash
# bash run.sh (testid0 testid1 ...)
# runs tests tests/testid0.sh tests/testid1.sh ...
# without arguments, it runs all tests
. tests/common.sh
TESTS=
if [ "$#" -eq 0 ]; then
for NAME in tests/??.sh; do
i=`basename $NAME .sh`
TESTS="$TESTS $i"
done
else
TESTS=$@
fi
ETH=../../ethereum
DIR="/tmp/eth.test/nodes"
TIMEOUT=10
mkdir -p $DIR/js
echo "running tests $TESTS"
for NAME in $TESTS; do
PIDS=
CHAIN="tests/$NAME.chain"
JSFILE="$DIR/js/$NAME.js"
CHAIN_TEST="$DIR/$NAME/chain"
echo "RUN: test $NAME"
cat tests/common.js > $JSFILE
. tests/$NAME.sh
sleep $TIMEOUT
echo "timeout after $TIMEOUT seconds: killing $PIDS"
kill $PIDS
if [ -r "$CHAIN" ]; then
if diff $CHAIN $CHAIN_TEST >/dev/null ; then
echo "chain ok: $CHAIN=$CHAIN_TEST"
else
echo "FAIL: chains differ: expected $CHAIN ; got $CHAIN_TEST"
continue
fi
fi
ERRORS=$DIR/errors
if [ -r "$ERRORS" ]; then
echo "FAIL: "
cat $ERRORS
else
echo PASS
fi
done
\ No newline at end of file
eth/test/tests/00.chain deleted 120000 → 0
View file @ cdc86965
../chains/01.chain
\ No newline at end of file
eth/test/tests/00.sh deleted 100644 → 0
View file @ cdc86965
#!/bin/bash
TIMEOUT=4
cat >> $JSFILE <<EOF
eth.addPeer("localhost:30311");
sleep(1000)
eth.export("$CHAIN_TEST");
EOF
peer 11 01
test_node $NAME "" -loglevel 5 $JSFILE
eth/test/tests/01.chain deleted 120000 → 0
View file @ cdc86965
../chains/02.chain
\ No newline at end of file
eth/test/tests/01.sh deleted 100644 → 0
View file @ cdc86965
#!/bin/bash
TIMEOUT=5
cat >> $JSFILE <<EOF
eth.addPeer("localhost:30311");
log("added peer localhost:30311");
sleep(1000);
log("added peer localhost:30312");
eth.addPeer("localhost:30312");
sleep(3000);
eth.export("$CHAIN_TEST");
EOF
peer 11 01
peer 12 02
test_node $NAME "" -loglevel 5 $JSFILE
eth/test/tests/02.chain deleted 120000 → 0
View file @ cdc86965
../chains/01.chain
\ No newline at end of file
eth/test/tests/02.sh deleted 100644 → 0
View file @ cdc86965
#!/bin/bash
TIMEOUT=6
cat >> $JSFILE <<EOF
eth.addPeer("localhost:30311");
sleep(200);
eth.addPeer("localhost:30312");
sleep(3000);
eth.export("$CHAIN_TEST");
EOF
peer 11 01
peer 12 02
P12ID=$PID
test_node $NAME "" -loglevel 5 $JSFILE
sleep 0.3
kill $P12ID
eth/test/tests/03.chain deleted 120000 → 0
View file @ cdc86965
../chains/12k.chain
\ No newline at end of file
eth/test/tests/03.sh deleted 100644 → 0
View file @ cdc86965
#!/bin/bash
TIMEOUT=12
cat >> $JSFILE <<EOF
eth.addPeer("localhost:30311");
sleep(10000);
eth.export("$CHAIN_TEST");
EOF
peer 11 12k
sleep 2
test_node $NAME "" -loglevel 5 $JSFILE
eth/test/tests/04.sh deleted 100644 → 0
View file @ cdc86965
#!/bin/bash
TIMEOUT=15
cat >> $JSFILE <<EOF
eth.addPeer("localhost:30311");
sleep(200);
eth.addPeer("localhost:30312");
sleep(13000);
eth.export("$CHAIN_TEST");
EOF
peer 11 01 -mine
peer 12 02
test_node $NAME "" -loglevel 5 $JSFILE
sleep 6
cat $DIR/$NAME/debug.log | grep 'best peer'
eth/test/tests/05.sh deleted 100644 → 0
View file @ cdc86965
#!/bin/bash
TIMEOUT=60
cat >> $JSFILE <<EOF
eth.addPeer("localhost:30311");
sleep(200);
eth.addPeer("localhost:30312");
eth.addPeer("localhost:30313");
eth.addPeer("localhost:30314");
sleep(3000);
eth.export("$CHAIN_TEST");
EOF
peer 11 01 -mine
peer 12 02 -mine
peer 13 03
peer 14 04
test_node $NAME "" -loglevel 5 $JSFILE
eth/test/tests/common.js deleted 100644 → 0
View file @ cdc86965
function log(text) {
console.log("[JS TEST SCRIPT] " + text);
}
function sleep(seconds) {
var now = new Date().getTime();
while(new Date().getTime() < now + seconds){}
}
eth/test/tests/common.sh deleted 100644 → 0
View file @ cdc86965
#!/bin/bash
# launched by run.sh
function test_node {
rm -rf $DIR/$1
ARGS="-datadir $DIR/$1 -debug debug -seed=false -shh=false -id test$1 -port 303$1"
if [ "" != "$2" ]; then
chain="chains/$2.chain"
echo "import chain $chain"
$ETH $ARGS -loglevel 3 -chain $chain | grep CLI |grep import
fi
echo "starting test node $1 with args $ARGS ${@:3}"
$ETH $ARGS ${@:3} &
PID=$!
PIDS="$PIDS $PID"
}
function peer {
test_node $@ -loglevel 5 -logfile debug.log -maxpeer 1 -dial=false
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment