aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorobscuren <geffobscura@gmail.com>2015-02-27 01:39:38 +0800
committerobscuren <geffobscura@gmail.com>2015-02-27 01:39:38 +0800
commitc18ea4de147cb81bf5563a5727172d4103658b92 (patch)
tree09ecb2fe54902855fcef4c1873d64cce4e28f399
parent37e6870f64437a212fde383ab1538ad1e7e2acd9 (diff)
parent16ecda951b767800b4e09ad8e86e0866b05136be (diff)
downloaddexon-c18ea4de147cb81bf5563a5727172d4103658b92.tar.gz
dexon-c18ea4de147cb81bf5563a5727172d4103658b92.tar.zst
dexon-c18ea4de147cb81bf5563a5727172d4103658b92.zip
Merge branch 'blockpool2' of https://github.com/ethersphere/go-ethereum into ethersphere-blockpool2
-rw-r--r--blockpool/blockpool.go749
-rw-r--r--blockpool/blockpool_test.go479
-rw-r--r--blockpool/blockpool_util_test.go350
-rw-r--r--blockpool/config_test.go40
-rw-r--r--blockpool/errors_test.go124
-rw-r--r--blockpool/peers.go536
-rw-r--r--blockpool/peers_test.go120
-rw-r--r--blockpool/section.go677
-rw-r--r--blockpool/status.go110
-rw-r--r--blockpool/status_test.go228
-rw-r--r--blockpool/test/hash_pool.go57
-rw-r--r--blockpool/test/logger.go78
-rw-r--r--blockpool/test/util.go35
-rw-r--r--errs/errors.go91
-rw-r--r--errs/errors_test.go46
-rw-r--r--eth/backend.go7
-rw-r--r--eth/block_pool.go1239
-rw-r--r--eth/block_pool_test.go983
-rw-r--r--eth/error.go72
-rw-r--r--eth/protocol.go69
-rw-r--r--eth/protocol_test.go18
-rw-r--r--eth/test/README.md27
-rw-r--r--eth/test/bootstrap.sh9
-rwxr-xr-xeth/test/chains/00.chainbin9726 -> 0 bytes
-rwxr-xr-xeth/test/chains/01.chainbin13881 -> 0 bytes
-rwxr-xr-xeth/test/chains/02.chainbin14989 -> 0 bytes
-rwxr-xr-xeth/test/chains/03.chainbin18590 -> 0 bytes
-rwxr-xr-xeth/test/chains/04.chainbin20529 -> 0 bytes
-rw-r--r--eth/test/mine.sh20
-rw-r--r--eth/test/run.sh53
l---------eth/test/tests/00.chain1
-rw-r--r--eth/test/tests/00.sh13
l---------eth/test/tests/01.chain1
-rw-r--r--eth/test/tests/01.sh18
l---------eth/test/tests/02.chain1
-rw-r--r--eth/test/tests/02.sh19
l---------eth/test/tests/03.chain1
-rw-r--r--eth/test/tests/03.sh14
-rw-r--r--eth/test/tests/04.sh17
-rw-r--r--eth/test/tests/05.sh20
-rw-r--r--eth/test/tests/common.js9
-rw-r--r--eth/test/tests/common.sh20
-rw-r--r--logger/loggers.go28
43 files changed, 3795 insertions, 2584 deletions
diff --git a/blockpool/blockpool.go b/blockpool/blockpool.go
new file mode 100644
index 000000000..0126d734c
--- /dev/null
+++ b/blockpool/blockpool.go
@@ -0,0 +1,749 @@
+package blockpool
+
+import (
+ "bytes"
+ "fmt"
+ "math/big"
+ "sync"
+ "time"
+
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/errs"
+ ethlogger "github.com/ethereum/go-ethereum/logger"
+ "github.com/ethereum/go-ethereum/pow"
+)
+
+var plog = ethlogger.NewLogger("Blockpool")
+
+var (
+ // max number of block hashes sent in one request
+ blockHashesBatchSize = 512
+ // max number of blocks sent in one request
+ blockBatchSize = 64
+ // interval between two consecutive block checks (and requests)
+ blocksRequestInterval = 3 * time.Second
+ // level of redundancy in block requests sent
+ blocksRequestRepetition = 1
+ // interval between two consecutive block hash checks (and requests)
+ blockHashesRequestInterval = 3 * time.Second
+ // max number of idle iterations, ie., check through a section without new blocks coming in
+ blocksRequestMaxIdleRounds = 100
+ // timeout interval: max time allowed for peer without sending a block hash
+ blockHashesTimeout = 60 * time.Second
+ // timeout interval: max time allowed for peer without sending a block
+ blocksTimeout = 120 * time.Second
+)
+
+// config embedded in components, by default fall back to constants
+// by default all resolved to local
+type Config struct {
+ BlockHashesBatchSize int
+ BlockBatchSize int
+ BlocksRequestRepetition int
+ BlocksRequestMaxIdleRounds int
+ BlockHashesRequestInterval time.Duration
+ BlocksRequestInterval time.Duration
+ BlockHashesTimeout time.Duration
+ BlocksTimeout time.Duration
+}
+
+// blockpool errors
+const (
+ ErrInvalidBlock = iota
+ ErrInvalidPoW
+ ErrUnrequestedBlock
+ ErrInsufficientChainInfo
+)
+
+var errorToString = map[int]string{
+ ErrInvalidBlock: "Invalid block",
+ ErrInvalidPoW: "Invalid PoW",
+ ErrUnrequestedBlock: "Unrequested block",
+ ErrInsufficientChainInfo: "Insufficient chain info",
+}
+
+// init initialises all your laundry
+func (self *Config) init() {
+ if self.BlockHashesBatchSize == 0 {
+ self.BlockHashesBatchSize = blockHashesBatchSize
+ }
+ if self.BlockBatchSize == 0 {
+ self.BlockBatchSize = blockBatchSize
+ }
+ if self.BlocksRequestRepetition == 0 {
+ self.BlocksRequestRepetition = blocksRequestRepetition
+ }
+ if self.BlocksRequestMaxIdleRounds == 0 {
+ self.BlocksRequestMaxIdleRounds = blocksRequestMaxIdleRounds
+ }
+ if self.BlockHashesRequestInterval == 0 {
+ self.BlockHashesRequestInterval = blockHashesRequestInterval
+ }
+ if self.BlocksRequestInterval == 0 {
+ self.BlocksRequestInterval = blocksRequestInterval
+ }
+ if self.BlockHashesTimeout == 0 {
+ self.BlockHashesTimeout = blockHashesTimeout
+ }
+ if self.BlocksTimeout == 0 {
+ self.BlocksTimeout = blocksTimeout
+ }
+}
+
+// node is the basic unit of the internal model of block chain/tree in the blockpool
+type node struct {
+ lock sync.RWMutex
+ hash []byte
+ block *types.Block
+ hashBy string
+ blockBy string
+ td *big.Int
+}
+
+type index struct {
+ int
+}
+
+// entry is the struct kept and indexed in the pool
+type entry struct {
+ node *node
+ section *section
+ index *index
+}
+
+type BlockPool struct {
+ Config *Config
+
+ // the minimal interface with blockchain
+ hasBlock func(hash []byte) bool
+ insertChain func(types.Blocks) error
+ verifyPoW func(pow.Block) bool
+
+ pool map[string]*entry
+ peers *peers
+
+ lock sync.RWMutex
+ chainLock sync.RWMutex
+ // alloc-easy pool of hash slices
+ hashSlicePool chan [][]byte
+
+ status *status
+
+ quit chan bool
+ wg sync.WaitGroup
+ running bool
+}
+
+// public constructor
+func New(
+ hasBlock func(hash []byte) bool,
+ insertChain func(types.Blocks) error,
+ verifyPoW func(pow.Block) bool,
+) *BlockPool {
+
+ return &BlockPool{
+ Config: &Config{},
+ hasBlock: hasBlock,
+ insertChain: insertChain,
+ verifyPoW: verifyPoW,
+ }
+}
+
+// allows restart
+func (self *BlockPool) Start() {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+
+ if self.running {
+ return
+ }
+
+ self.Config.init()
+ self.hashSlicePool = make(chan [][]byte, 150)
+ self.status = newStatus()
+ self.quit = make(chan bool)
+ self.pool = make(map[string]*entry)
+ self.running = true
+
+ self.peers = &peers{
+ errors: &errs.Errors{
+ Package: "Blockpool",
+ Errors: errorToString,
+ },
+ peers: make(map[string]*peer),
+ status: self.status,
+ bp: self,
+ }
+ timer := time.NewTicker(3 * time.Second)
+ go func() {
+ for {
+ select {
+ case <-self.quit:
+ return
+ case <-timer.C:
+ plog.Debugf("status:\n%v", self.Status())
+ }
+ }
+ }()
+ plog.Infoln("Started")
+}
+
+func (self *BlockPool) Stop() {
+ self.lock.Lock()
+ if !self.running {
+ self.lock.Unlock()
+ return
+ }
+ self.running = false
+
+ self.lock.Unlock()
+
+ plog.Infoln("Stopping...")
+
+ close(self.quit)
+
+ self.lock.Lock()
+ self.peers = nil
+ self.pool = nil
+ self.lock.Unlock()
+
+ plog.Infoln("Stopped")
+}
+
+// Wait blocks until active processes finish
+func (self *BlockPool) Wait(t time.Duration) {
+ self.lock.Lock()
+ if !self.running {
+ self.lock.Unlock()
+ return
+ }
+ self.lock.Unlock()
+
+ plog.Infoln("Waiting for processes to complete...")
+ w := make(chan bool)
+ go func() {
+ self.wg.Wait()
+ close(w)
+ }()
+
+ select {
+ case <-w:
+ plog.Infoln("Processes complete")
+ case <-time.After(t):
+ plog.Warnf("Timeout")
+ }
+}
+
+/*
+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
+Called a second time with the same peer id, it is used to update chain info for a peer. This is used when a new (mined) block message is received.
+RemovePeer needs to be called when the peer disconnects.
+Peer info is currently not persisted across disconnects (or sessions)
+*/
+func (self *BlockPool) AddPeer(
+
+ td *big.Int, currentBlockHash []byte,
+ peerId string,
+ requestBlockHashes func([]byte) error,
+ requestBlocks func([][]byte) error,
+ peerError func(*errs.Error),
+
+) (best bool) {
+
+ return self.peers.addPeer(td, currentBlockHash, peerId, requestBlockHashes, requestBlocks, peerError)
+}
+
+// RemovePeer needs to be called when the peer disconnects
+func (self *BlockPool) RemovePeer(peerId string) {
+ self.peers.removePeer(peerId)
+}
+
+/*
+AddBlockHashes
+
+Entry point for eth protocol to add block hashes received via BlockHashesMsg
+
+only hashes from the best peer are handled
+
+initiates further hash requests until a known parent is reached (unless cancelled by a peerSwitch event, i.e., when a better peer becomes best peer)
+launches all block request processes on each chain section
+
+the first argument is an iterator function. Using this block hashes are decoded from the rlp message payload on demand. As a result, AddBlockHashes needs to run synchronously for one peer since the message is discarded if the caller thread returns.
+*/
+func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) {
+
+ bestpeer, best := self.peers.getPeer(peerId)
+ if !best {
+ return
+ }
+ // bestpeer is still the best peer
+
+ self.wg.Add(1)
+
+ defer func() { self.wg.Done() }()
+
+ self.status.lock.Lock()
+ self.status.activePeers[bestpeer.id]++
+ self.status.lock.Unlock()
+
+ var n int
+ var hash []byte
+ var ok, headSection, peerswitch bool
+ var sec, child, parent *section
+ var entry *entry
+ var nodes []*node
+
+ hash, ok = next()
+ bestpeer.lock.Lock()
+
+ plog.Debugf("AddBlockHashes: peer <%s> starting from [%s] (peer head: %s)", peerId, hex(bestpeer.parentHash), hex(bestpeer.currentBlockHash))
+
+ // first check if we are building the head section of a peer's chain
+ if bytes.Equal(bestpeer.parentHash, hash) {
+ if self.hasBlock(bestpeer.currentBlockHash) {
+ return
+ }
+ /*
+ when peer is promoted in switchPeer, a new header section process is launched
+ as the head section skeleton is actually created here, it is signaled to the process
+ so that it can quit
+ in the special case that the node for parent of the head block is found in the blockpool
+ (with or without fetched block)
+ */
+ headSection = true
+ if entry := self.get(bestpeer.currentBlockHash); entry == nil {
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) head section starting from [%s] ", peerId, hex(bestpeer.currentBlockHash), hex(bestpeer.parentHash))
+ // if head block is not yet in the pool, create entry and start node list for section
+ node := &node{
+ hash: bestpeer.currentBlockHash,
+ block: bestpeer.currentBlock,
+ hashBy: peerId,
+ blockBy: peerId,
+ }
+ // nodes is a list of nodes in one section ordered top-bottom (old to young)
+ nodes = append(nodes, node)
+ n++
+ } else {
+ // otherwise set child section iff found node is the root of a section
+ // this is a possible scenario when a singleton head section was created
+ // on an earlier occasion this peer or another with the same block was best peer
+ if entry.node == entry.section.bottom {
+ child = entry.section
+ plog.DebugDetailf("AddBlockHashes: peer <%s>: connects to child section root %s", peerId, hex(bestpeer.currentBlockHash))
+ }
+ }
+ } else {
+ // otherwise : we are not building the head section of the peer
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) section starting from [%s] ", peerId, hex(bestpeer.currentBlockHash), hex(hash))
+ }
+ // the switch channel signals peerswitch event
+ switchC := bestpeer.switchC
+ bestpeer.lock.Unlock()
+
+ // iterate over hashes coming from peer (first round we have hash set above)
+LOOP:
+ for ; ok; hash, ok = next() {
+
+ select {
+ case <-self.quit:
+ // global quit for blockpool
+ return
+
+ case <-switchC:
+ // if the peer is demoted, no more hashes read
+ plog.DebugDetailf("AddBlockHashes: demoted peer <%s> (head: %s)", peerId, hex(bestpeer.currentBlockHash), hex(hash))
+ peerswitch = true
+ break LOOP
+ default:
+ }
+
+ // if we reach the blockchain we stop reading more
+ if self.hasBlock(hash) {
+ // check if known block connecting the downloaded chain to our blockchain
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) found block %s in the blockchain", peerId, hex(bestpeer.currentBlockHash), hex(hash))
+ if len(nodes) == 1 {
+ // create new section if needed and push it to the blockchain
+ sec = self.newSection(nodes)
+ sec.addSectionToBlockChain(bestpeer)
+ } else {
+
+ /*
+ not added hash yet but according to peer child section built
+ earlier chain connects with blockchain
+ this maybe a potential vulnarability
+ the root block arrives (or already there but its parenthash was not pointing to known block in the blockchain)
+ we start inserting -> error -> remove the entire chain
+ instead of punishing this peer
+ solution: when switching peers always make sure best peers own head block
+ and td together with blockBy are recorded on the node
+ */
+ if len(nodes) == 0 && child != nil {
+ child.addSectionToBlockChain(bestpeer)
+ }
+ }
+ break LOOP
+ }
+
+ // look up node in the 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
+ this also supports clients that (despite the spec) include <from> hash in their
+ response to hashes request. Note that by providing <from> we can link sections
+ without having to wait for the root block of the child section to arrive, so it allows for superior performance
+ */
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) found head block [%s] as root of connecting child section [%s] skipping", peerId, hex(bestpeer.currentBlockHash), hex(hash), sectionhex(entry.section))
+ // record the entry's chain section as child section
+ child = entry.section
+ continue LOOP
+ }
+ // otherwise record entry's chain section as parent connecting it to the pool
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) found block [%s] in section [%s]. Connected to pool.", peerId, hex(bestpeer.currentBlockHash), hex(hash), sectionhex(entry.section))
+ parent = entry.section
+ break LOOP
+ }
+
+ // finally if node for block hash does not exist, create it and append node to section nodes
+ node := &node{
+ hash: hash,
+ hashBy: peerId,
+ }
+ nodes = append(nodes, node)
+ } //for
+
+ /*
+ we got here if
+ - run out of hashes (parent = nil) sent by our best peer
+ - our peer is demoted (peerswitch = true)
+ - reached blockchain or blockpool
+ - quitting
+ */
+ self.chainLock.Lock()
+
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): %v nodes in new section", peerId, hex(bestpeer.currentBlockHash), len(nodes))
+ /*
+ handle forks where connecting node is mid-section
+ by splitting section at fork
+ no splitting needed if connecting node is head of a section
+ */
+ if parent != nil && entry != nil && entry.node != parent.top && len(nodes) > 0 {
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): fork after %s", peerId, hex(bestpeer.currentBlockHash), hex(hash))
+
+ self.splitSection(parent, entry)
+
+ self.status.lock.Lock()
+ self.status.values.Forks++
+ self.status.lock.Unlock()
+ }
+
+ /*
+ if new section is created, link it to parent/child sections
+ and launch section process fetching blocks and further hashes
+ */
+ sec = self.linkSections(nodes, parent, child)
+
+ self.status.lock.Lock()
+ self.status.values.BlockHashes += len(nodes)
+ self.status.lock.Unlock()
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): section [%s] created", peerId, hex(bestpeer.currentBlockHash), sectionhex(sec))
+
+ self.chainLock.Unlock()
+
+ /*
+ if a blockpool node is reached (parent section is not nil),
+ activate section (unless our peer is demoted by now).
+ this can be the bottom half of a newly split section in case of a fork.
+ bestPeer is nil if we got here after our peer got demoted while processing.
+ in this case no activation should happen
+ */
+ if parent != nil && !peerswitch {
+ self.activateChain(parent, bestpeer, nil)
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): parent section [%s]", peerId, hex(bestpeer.currentBlockHash), sectionhex(parent))
+ }
+
+ /*
+ if a new section was created,
+ register section iff head section or no child known
+ activate it with this peer
+ */
+ if sec != nil {
+ // switch on section process (it is paused by switchC)
+ if !peerswitch {
+ if headSection || child == nil {
+ bestpeer.lock.Lock()
+ bestpeer.sections = append(bestpeer.sections, sec.top.hash)
+ bestpeer.lock.Unlock()
+ }
+ /*
+ request next block hashes for parent section here.
+ but only once, repeating only when bottom block arrives,
+ otherwise no way to check if it arrived
+ */
+ bestpeer.requestBlockHashes(sec.bottom.hash)
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): start requesting blocks for section [%s]", peerId, hex(bestpeer.currentBlockHash), sectionhex(sec))
+ sec.activate(bestpeer)
+ } else {
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) no longer best: delay requesting blocks for section [%s]", peerId, hex(bestpeer.currentBlockHash), sectionhex(sec))
+ sec.deactivate()
+ }
+ }
+
+ // if we are processing peer's head section, signal it to headSection process that it is created
+
+ if headSection {
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) head section registered on head section process", peerId, hex(bestpeer.currentBlockHash))
+
+ var headSec *section
+ switch {
+ case sec != nil:
+ headSec = sec
+ case child != nil:
+ headSec = child
+ default:
+ headSec = parent
+ }
+ if !peerswitch {
+ plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) head section [%s] created signalled to head section process", peerId, hex(bestpeer.currentBlockHash), sectionhex(headSec))
+ bestpeer.headSectionC <- headSec
+ }
+ }
+}
+
+/*
+ AddBlock is the entry point for the eth protocol to call when blockMsg is received.
+
+ It has a strict interpretation of the protocol in that if the block received has not been requested, it results in an error
+
+ At the same time it is opportunistic in that if a requested block may be provided by any peer.
+
+ The received 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()
+
+ sender, _ := self.peers.getPeer(peerId)
+ if sender == nil {
+ return
+ }
+
+ self.status.lock.Lock()
+ self.status.activePeers[peerId]++
+ self.status.lock.Unlock()
+
+ entry := self.get(hash)
+
+ // a peer's current head block is appearing the first time
+ if bytes.Equal(hash, sender.currentBlockHash) {
+ if sender.currentBlock == nil {
+ plog.Debugf("AddBlock: add head block %s for peer <%s> (head: %s)", hex(hash), peerId, hex(sender.currentBlockHash))
+ sender.setChainInfoFromBlock(block)
+ // sender.currentBlockC <- block
+
+ self.status.lock.Lock()
+ self.status.values.Blocks++
+ self.status.values.BlocksInPool++
+ self.status.lock.Unlock()
+ } else {
+ plog.DebugDetailf("AddBlock: head block %s for peer <%s> (head: %s) already known", hex(hash), peerId, hex(sender.currentBlockHash))
+ }
+ } else {
+
+ plog.DebugDetailf("AddBlock: block %s received from peer <%s> (head: %s)", hex(hash), peerId, hex(sender.currentBlockHash))
+
+ sender.lock.Lock()
+ // update peer chain info if more recent than what we registered
+ if block.Td != nil && block.Td.Cmp(sender.td) > 0 {
+ sender.td = block.Td
+ sender.currentBlockHash = block.Hash()
+ sender.parentHash = block.ParentHash()
+ sender.currentBlock = block
+ sender.headSection = nil
+ }
+ sender.lock.Unlock()
+
+ if entry == nil {
+ // penalise peer for sending what we have not asked
+ plog.DebugDetailf("AddBlock: unrequested block %s received from peer <%s> (head: %s)", hex(hash), peerId, hex(sender.currentBlockHash))
+ sender.addError(ErrUnrequestedBlock, "%x", hash)
+
+ self.status.lock.Lock()
+ self.status.badPeers[peerId]++
+ self.status.lock.Unlock()
+ return
+ }
+ }
+ if entry == nil {
+ return
+ }
+
+ node := entry.node
+ node.lock.Lock()
+ defer node.lock.Unlock()
+
+ // check if block already present
+ if node.block != nil {
+ plog.DebugDetailf("AddBlock: block %s from peer <%s> (head: %s) already sent by <%s> ", hex(hash), peerId, hex(sender.currentBlockHash), node.blockBy)
+ return
+ }
+
+ // check if block is already inserted in the blockchain
+ if self.hasBlock(hash) {
+ plog.DebugDetailf("AddBlock: block %s from peer <%s> (head: %s) already in the blockchain", hex(hash), peerId, hex(sender.currentBlockHash))
+ return
+ }
+
+ // validate block for PoW
+ if !self.verifyPoW(block) {
+ plog.Warnf("AddBlock: invalid PoW on block %s from peer <%s> (head: %s)", hex(hash), peerId, hex(sender.currentBlockHash))
+ sender.addError(ErrInvalidPoW, "%x", hash)
+
+ self.status.lock.Lock()
+ self.status.badPeers[peerId]++
+ self.status.lock.Unlock()
+
+ return
+ }
+
+ node.block = block
+ node.blockBy = peerId
+ node.td = block.Td // optional field
+
+ self.status.lock.Lock()
+ self.status.values.Blocks++
+ self.status.values.BlocksInPool++
+ self.status.lock.Unlock()
+
+}
+
+/*
+ iterates down a chain section by section
+ activating section process on incomplete sections with peer
+ relinking orphaned sections with their parent if root block (and its parent hash) is known)
+*/
+func (self *BlockPool) activateChain(sec *section, p *peer, connected map[string]*section) {
+
+ p.lock.RLock()
+ switchC := p.switchC
+ p.lock.RUnlock()
+
+ var i int
+
+LOOP:
+ for sec != nil {
+ parent := self.getParent(sec)
+ plog.DebugDetailf("activateChain: section [%s] activated by peer <%s>", sectionhex(sec), p.id)
+ sec.activate(p)
+ if i > 0 && connected != nil {
+ connected[string(sec.top.hash)] = sec
+ }
+ /*
+ we need to relink both complete and incomplete sections
+ the latter could have been blockHashesRequestsComplete before being delinked from its parent
+ */
+ if parent == nil && sec.bottom.block != nil {
+ if entry := self.get(sec.bottom.block.ParentHash()); entry != nil {
+ parent = entry.section
+ plog.DebugDetailf("activateChain: [%s]-[%s] relink", sectionhex(parent), sectionhex(sec))
+ link(parent, sec)
+ }
+ }
+ sec = parent
+
+ // stop if peer got demoted
+ select {
+ case <-switchC:
+ break LOOP
+ case <-self.quit:
+ break LOOP
+ default:
+ }
+ }
+}
+
+// must run in separate go routine, otherwise
+// switchpeer -> activateChain -> activate deadlocks on section process select and peers.lock
+func (self *BlockPool) requestBlocks(attempts int, hashes [][]byte) {
+ self.wg.Add(1)
+ go func() {
+ self.peers.requestBlocks(attempts, hashes)
+ self.wg.Done()
+ }()
+}
+
+// convenience methods to access adjacent sections
+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
+}
+
+// accessor and setter for entries in the pool
+func (self *BlockPool) get(hash []byte) *entry {
+ self.lock.RLock()
+ defer self.lock.RUnlock()
+ return self.pool[string(hash)]
+}
+
+func (self *BlockPool) set(hash []byte, e *entry) {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ self.pool[string(hash)] = e
+}
+
+func (self *BlockPool) remove(sec *section) {
+ // delete node entries from pool index under pool lock
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ for _, node := range sec.nodes {
+ delete(self.pool, string(node.hash))
+ }
+}
+
+func (self *BlockPool) getHashSlice() (s [][]byte) {
+ select {
+ case s = <-self.hashSlicePool:
+ default:
+ s = make([][]byte, self.Config.BlockBatchSize)
+ }
+ return
+}
+
+// Return returns a Client to the pool.
+func (self *BlockPool) putHashSlice(s [][]byte) {
+ if len(s) == self.Config.BlockBatchSize {
+ select {
+ case self.hashSlicePool <- s:
+ default:
+ }
+ }
+}
+
+// pretty prints hash (byte array) with first 4 bytes in hex
+func hex(hash []byte) (name string) {
+ if hash == nil {
+ name = ""
+ } else {
+ name = fmt.Sprintf("%x", hash[:4])
+ }
+ return
+}
+
+// pretty prints a section using first 4 bytes in hex of bottom and top blockhash of the section
+func sectionhex(section *section) (name string) {
+ if section == nil {
+ name = ""
+ } else {
+ name = fmt.Sprintf("%x-%x", section.bottom.hash[:4], section.top.hash[:4])
+ }
+ return
+}
diff --git a/blockpool/blockpool_test.go b/blockpool/blockpool_test.go
new file mode 100644
index 000000000..bca48c6ca
--- /dev/null
+++ b/blockpool/blockpool_test.go
@@ -0,0 +1,479 @@
+package blockpool
+
+import (
+ "testing"
+ "time"
+
+ "github.com/ethereum/go-ethereum/blockpool/test"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/ethutil"
+)
+
+func TestPeerWithKnownBlock(t *testing.T) {
+ test.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)
+ blockPool.Stop()
+ // no request on known block
+ peer0.checkBlockHashesRequests()
+}
+
+func TestPeerWithKnownParentBlock(t *testing.T) {
+ test.LogInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.initRefBlockChain(1)
+ blockPoolTester.blockChain[0] = nil
+ blockPool.Start()
+
+ peer0 := blockPoolTester.newPeer("0", 1, 1)
+ peer0.AddPeer()
+ peer0.serveBlocks(0, 1)
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ peer0.checkBlocksRequests([]int{1})
+ peer0.checkBlockHashesRequests()
+ blockPoolTester.refBlockChain[1] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestPeerPromotionByOptionalTdOnBlock(t *testing.T) {
+ test.LogInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(4)
+ peer0 := blockPoolTester.newPeer("peer0", 2, 2)
+ peer1 := blockPoolTester.newPeer("peer1", 1, 1)
+ peer2 := blockPoolTester.newPeer("peer2", 3, 4)
+
+ blockPool.Start()
+
+ // pool
+ peer0.AddPeer()
+ peer0.serveBlocks(1, 2)
+ best := peer1.AddPeer()
+ // this tests that peer1 is not promoted over peer0 yet
+ if best {
+ t.Errorf("peer1 (TD=1) should not be set as best")
+ }
+ best = peer2.AddPeer()
+ peer2.serveBlocks(3, 4)
+ peer2.serveBlockHashes(4, 3, 2, 1)
+ hashes := blockPoolTester.hashPool.IndexesToHashes([]int{2, 3})
+ peer1.waitBlocksRequests(3)
+ blockPool.AddBlock(&types.Block{
+ HeaderHash: ethutil.Bytes(hashes[1]),
+ ParentHeaderHash: ethutil.Bytes(hashes[0]),
+ Td: ethutil.Big3,
+ }, "peer1")
+
+ blockPool.RemovePeer("peer2")
+ if blockPool.peers.best.id != "peer1" {
+ t.Errorf("peer1 (TD=3) should be set as best")
+ }
+ peer1.serveBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[4] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestSimpleChain(t *testing.T) {
+ test.LogInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(2)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 2)
+ peer1.AddPeer()
+ peer1.serveBlocks(1, 2)
+ go peer1.serveBlockHashes(2, 1, 0)
+ peer1.serveBlocks(0, 1)
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[2] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestChainConnectingWithParentHash(t *testing.T) {
+ test.LogInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(3)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 3)
+ peer1.AddPeer()
+ go peer1.serveBlocks(2, 3)
+ go peer1.serveBlockHashes(3, 2, 1)
+ peer1.serveBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[3] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestMultiSectionChain(t *testing.T) {
+ test.LogInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(5)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 5)
+
+ peer1.AddPeer()
+ go peer1.serveBlocks(4, 5)
+ go peer1.serveBlockHashes(5, 4, 3)
+ go peer1.serveBlocks(2, 3, 4)
+ go peer1.serveBlockHashes(3, 2, 1, 0)
+ peer1.serveBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[5] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestNewBlocksOnPartialChain(t *testing.T) {
+ test.LogInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(7)
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 5)
+
+ peer1.AddPeer()
+ go peer1.serveBlocks(4, 5) // partially complete section
+ go peer1.serveBlockHashes(5, 4, 3)
+ peer1.serveBlocks(3, 4) // partially complete section
+ // peer1 found new blocks
+ peer1.td = 2
+ peer1.currentBlock = 7
+ peer1.AddPeer()
+ peer1.sendBlocks(6, 7)
+ go peer1.serveBlockHashes(7, 6, 5)
+ go peer1.serveBlocks(2, 3)
+ go peer1.serveBlocks(5, 6)
+ go peer1.serveBlockHashes(3, 2, 1) // tests that hash request from known chain root is remembered
+ peer1.serveBlocks(0, 1, 2)
+ // blockPool.RemovePeer("peer1")
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[7] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestPeerSwitchUp(t *testing.T) {
+ test.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)
+
+ peer1.AddPeer()
+ go peer1.serveBlocks(5, 6)
+ go peer1.serveBlockHashes(6, 5, 4, 3) //
+ peer1.serveBlocks(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.serveBlocks(6, 7)
+ // go peer2.serveBlockHashes(7, 6) //
+ go peer2.serveBlocks(4, 5) // tests that block request for earlier section is remembered
+ go peer1.serveBlocks(3, 4) // tests that connecting section by demoted peer is remembered and blocks are accepted from demoted peer
+ go peer2.serveBlockHashes(3, 2, 1, 0) // tests that known chain section is activated, hash requests from 3 is remembered
+ peer2.serveBlocks(0, 1, 2) // final blocks linking to blockchain sent
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[7] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestPeerSwitchDownOverlapSectionWithoutRootBlock(t *testing.T) {
+ test.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.AddPeer()
+ peer2.serveBlocks(5, 6) // partially complete, section will be preserved
+ peer2.serveBlockHashes(6, 5, 4) // no go: make sure skeleton is created
+ peer1.AddPeer() // inferior peer1 is promoted as best peer
+ blockPool.RemovePeer("peer2") // peer2 disconnects
+ go peer1.serveBlockHashes(4, 3, 2, 1, 0) //
+ go peer1.serveBlocks(3, 4) //
+ go peer1.serveBlocks(4, 5) // tests that section set by demoted peer is remembered and blocks are accepted from new peer if they have it even if peers original TD is lower
+ peer1.serveBlocks(0, 1, 2, 3)
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[6] = []int{} // tests that idle sections are not inserted in blockchain
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestPeerSwitchDownOverlapSectionWithRootBlock(t *testing.T) {
+ test.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.AddPeer()
+ peer2.serveBlocks(5, 6) // partially complete, section will be preserved
+ go peer2.serveBlockHashes(6, 5, 4) //
+ peer2.serveBlocks(3, 4) // !incomplete section
+ time.Sleep(100 * time.Millisecond) // make sure block 4 added
+ peer1.AddPeer() // inferior peer1 is promoted as best peer
+ blockPool.RemovePeer("peer2") // peer2 disconnects
+ go peer1.serveBlockHashes(4, 3, 2, 1, 0) // tests that hash request are directly connecting if the head block exists
+ go peer1.serveBlocks(4, 5) // tests that section set by demoted peer is remembered and blocks are accepted from new peer if they have it even if peers original TD is lower
+ peer1.serveBlocks(0, 1, 2, 3)
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[6] = []int{} // tests that idle sections are not inserted in blockchain
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestPeerSwitchDownDisjointSection(t *testing.T) {
+ test.LogInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(3)
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 3)
+ peer2 := blockPoolTester.newPeer("peer2", 2, 6)
+
+ peer2.AddPeer()
+ peer2.serveBlocks(5, 6) // partially complete, section will be preserved
+ go peer2.serveBlockHashes(6, 5, 4) //
+ peer2.serveBlocks(3, 4, 5) //
+ time.Sleep(100 * time.Millisecond) // make sure blocks are received
+ peer1.AddPeer() // inferior peer1 is promoted as best peer
+ blockPool.RemovePeer("peer2") // peer2 disconnects
+ go peer1.serveBlocks(2, 3) //
+ go peer1.serveBlockHashes(3, 2, 1) //
+ peer1.serveBlocks(0, 1, 2) //
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[3] = []int{} // tests that idle sections are not inserted in blockchain
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestPeerSwitchBack(t *testing.T) {
+ test.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.AddPeer()
+ go peer2.serveBlocks(7, 8)
+ go peer2.serveBlockHashes(8, 7, 6)
+ go peer2.serveBlockHashes(6, 5, 4)
+ peer2.serveBlocks(4, 5) // section partially complete
+ peer1.AddPeer() // peer1 is promoted as best peer
+ go peer1.serveBlocks(10, 11) //
+ peer1.serveBlockHashes(11, 10) // only gives useless results
+ blockPool.RemovePeer("peer1") // peer1 disconnects
+ go peer2.serveBlockHashes(4, 3, 2, 1, 0) // tests that asking for hashes from 4 is remembered
+ go peer2.serveBlocks(3, 4, 5, 6, 7, 8) // tests that section 4, 5, 6 and 7, 8 are remembered for missing blocks
+ peer2.serveBlocks(0, 1, 2, 3)
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[8] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestForkSimple(t *testing.T) {
+ test.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)
+
+ peer1.AddPeer()
+ go peer1.serveBlocks(8, 9)
+ go peer1.serveBlockHashes(9, 8, 7, 3, 2)
+ peer1.serveBlocks(1, 2, 3, 7, 8)
+ peer2.AddPeer() // peer2 is promoted as best peer
+ go peer2.serveBlocks(5, 6) //
+ go peer2.serveBlockHashes(6, 5, 4, 3, 2) // fork on 3 -> 4 (earlier child: 7)
+ go peer2.serveBlocks(1, 2, 3, 4, 5)
+ go peer2.serveBlockHashes(2, 1, 0)
+ peer2.serveBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout)
+ 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) {
+ test.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)
+
+ peer1.AddPeer()
+ go peer1.serveBlocks(8, 9) //
+ go peer1.serveBlockHashes(9, 8, 7, 3, 2) //
+ peer1.serveBlocks(7, 8) // partial section
+ // time.Sleep(1 * time.Second)
+ peer2.AddPeer() //
+ go peer2.serveBlocks(5, 6) //
+ go peer2.serveBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
+ peer2.serveBlocks(1, 2, 3, 4, 5) //
+
+ // peer1 finds new blocks
+ peer1.td = 3
+ peer1.currentBlock = 11
+ peer1.AddPeer()
+ go peer1.serveBlocks(10, 11)
+ go peer1.serveBlockHashes(11, 10, 9)
+ go peer1.serveBlocks(9, 10)
+ // time.Sleep(1 * time.Second)
+ go peer1.serveBlocks(3, 7) // tests that block requests on earlier fork are remembered
+ go peer1.serveBlockHashes(2, 1) // tests that hash request from root of connecting chain section (added by demoted peer) is remembered
+ peer1.serveBlocks(0, 1)
+
+ blockPool.Wait(waitTimeout)
+ 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) {
+ test.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)
+
+ peer1.AddPeer()
+ go peer1.serveBlocks(8, 9)
+ go peer1.serveBlockHashes(9, 8, 7, 3, 2)
+ peer1.serveBlocks(7, 8)
+ peer2.AddPeer()
+ go peer2.serveBlocks(5, 6) //
+ go peer2.serveBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
+ peer2.serveBlocks(2, 3, 4, 5) //
+ blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer
+ go peer1.serveBlocks(1, 2) //
+ go peer1.serveBlockHashes(2, 1, 0) //
+ go peer1.serveBlocks(3, 7) // tests that block requests on earlier fork are remembered and orphan section relinks to existing parent block
+ peer1.serveBlocks(0, 1)
+
+ blockPool.Wait(waitTimeout)
+ 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) {
+ test.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)
+
+ peer1.AddPeer()
+ go peer1.serveBlocks(8, 9)
+ go peer1.serveBlockHashes(9, 8, 7)
+ peer1.serveBlocks(3, 7, 8) // make sure this section is complete
+ time.Sleep(1 * time.Second)
+ go peer1.serveBlockHashes(7, 3, 2) // block 3/7 is section boundary
+ peer1.serveBlocks(2, 3) // partially complete sections block 2 missing
+ peer2.AddPeer() //
+ go peer2.serveBlocks(5, 6) //
+ go peer2.serveBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
+ peer2.serveBlocks(2, 3, 4, 5) // block 2 still missing.
+ blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer
+ // peer1.serveBlockHashes(7, 3) // tests that hash request from fork root is remembered even though section process completed
+ go peer1.serveBlockHashes(2, 1, 0) //
+ peer1.serveBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout)
+ 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)
+
+}
diff --git a/blockpool/blockpool_util_test.go b/blockpool/blockpool_util_test.go
new file mode 100644
index 000000000..8f0c8baa0
--- /dev/null
+++ b/blockpool/blockpool_util_test.go
@@ -0,0 +1,350 @@
+package blockpool
+
+import (
+ "fmt"
+ "math/big"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/ethereum/go-ethereum/blockpool/test"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/errs"
+ "github.com/ethereum/go-ethereum/ethutil"
+ "github.com/ethereum/go-ethereum/pow"
+)
+
+var (
+ waitTimeout = 60 * time.Second
+ testBlockHashesRequestInterval = 10 * time.Millisecond
+ testBlocksRequestInterval = 10 * time.Millisecond
+ requestWatchInterval = 10 * time.Millisecond
+)
+
+// 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 *test.TestHashPool
+ lock sync.RWMutex
+ reqlock sync.RWMutex
+ blocksRequestsMap map[int]bool
+ refBlockChain blockChain
+ blockChain blockChain
+ blockPool *BlockPool
+ t *testing.T
+}
+
+func newTestBlockPool(t *testing.T) (hashPool *test.TestHashPool, blockPool *BlockPool, b *blockPoolTester) {
+ hashPool = test.NewHashPool()
+ b = &blockPoolTester{
+ t: t,
+ hashPool: hashPool,
+ blockChain: make(blockChain),
+ refBlockChain: make(blockChain),
+ blocksRequestsMap: make(map[int]bool),
+ }
+ b.blockPool = New(b.hasBlock, b.insertChain, b.verifyPoW)
+ blockPool = b.blockPool
+ blockPool.Config.BlockHashesRequestInterval = testBlockHashesRequestInterval
+ blockPool.Config.BlocksRequestInterval = testBlocksRequestInterval
+ 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.Lock()
+ defer self.lock.Unlock()
+ 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) {
+ self.lock.RLock()
+ defer self.lock.RUnlock()
+ 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 || !test.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 the 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 *test.TestHashPool
+ lock sync.RWMutex
+ bt *blockPoolTester
+ 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,
+ bt: self,
+ blocksRequestsMap: self.blocksRequestsMap,
+ }
+}
+
+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 !test.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 !test.ArrayEq(r, rr) {
+ self.Errorf("block hashes requests incorrect\ngot %v\nexpected %v", r, rr)
+ }
+ }
+}
+
+// waiter function used by peer.serveBlocks
+// blocking until requests appear
+// since block requests are sent to any random peers
+// block request map is shared between peers
+// times out after waitTimeout
+func (self *peerTester) waitBlocksRequests(blocksRequest ...int) {
+ timeout := time.After(waitTimeout)
+ 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(requestWatchInterval)
+ select {
+ case <-timeout:
+ default:
+ }
+ }
+}
+
+// waiter function used by peer.serveBlockHashes
+// blocking until requests appear
+// times out after a period
+func (self *peerTester) waitBlockHashesRequests(blocksHashesRequest int) {
+ timeout := time.After(waitTimeout)
+ 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(requestWatchInterval)
+ 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) serveBlockHashes(indexes ...int) {
+ fmt.Printf("ready to serve block hashes %v\n", indexes)
+
+ self.waitBlockHashesRequests(indexes[0])
+ self.sendBlockHashes(indexes...)
+}
+
+func (self *peerTester) sendBlockHashes(indexes ...int) {
+ 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) serveBlocks(indexes ...int) {
+ fmt.Printf("ready to serve blocks %v\n", indexes[1:])
+ self.waitBlocksRequests(indexes[1:]...)
+ self.sendBlocks(indexes...)
+}
+
+func (self *peerTester) sendBlocks(indexes ...int) {
+ fmt.Printf("adding blocks %v \n", indexes)
+ hashes := self.hashPool.IndexesToHashes(indexes)
+ 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] block 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.bt.reqlock.Lock()
+ defer self.bt.reqlock.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(err *errs.Error) {
+ self.peerErrors = append(self.peerErrors, err.Code)
+}
diff --git a/blockpool/config_test.go b/blockpool/config_test.go
new file mode 100644
index 000000000..c06649024
--- /dev/null
+++ b/blockpool/config_test.go
@@ -0,0 +1,40 @@
+package blockpool
+
+import (
+ "testing"
+ "time"
+
+ "github.com/ethereum/go-ethereum/blockpool/test"
+)
+
+func TestBlockPoolConfig(t *testing.T) {
+ test.LogInit()
+ blockPool := &BlockPool{Config: &Config{}}
+ blockPool.Start()
+ c := blockPool.Config
+ test.CheckInt("BlockHashesBatchSize", c.BlockHashesBatchSize, blockHashesBatchSize, t)
+ test.CheckInt("BlockBatchSize", c.BlockBatchSize, blockBatchSize, t)
+ test.CheckInt("BlocksRequestRepetition", c.BlocksRequestRepetition, blocksRequestRepetition, t)
+ test.CheckInt("BlocksRequestMaxIdleRounds", c.BlocksRequestMaxIdleRounds, blocksRequestMaxIdleRounds, t)
+ test.CheckDuration("BlockHashesRequestInterval", c.BlockHashesRequestInterval, blockHashesRequestInterval, t)
+ test.CheckDuration("BlocksRequestInterval", c.BlocksRequestInterval, blocksRequestInterval, t)
+ test.CheckDuration("BlockHashesTimeout", c.BlockHashesTimeout, blockHashesTimeout, t)
+ test.CheckDuration("BlocksTimeout", c.BlocksTimeout, blocksTimeout, t)
+}
+
+func TestBlockPoolOverrideConfig(t *testing.T) {
+ test.LogInit()
+ blockPool := &BlockPool{Config: &Config{}}
+ c := &Config{128, 32, 1, 0, 300 * time.Millisecond, 100 * time.Millisecond, 90 * time.Second, 0}
+
+ blockPool.Config = c
+ blockPool.Start()
+ test.CheckInt("BlockHashesBatchSize", c.BlockHashesBatchSize, 128, t)
+ test.CheckInt("BlockBatchSize", c.BlockBatchSize, 32, t)
+ test.CheckInt("BlocksRequestRepetition", c.BlocksRequestRepetition, blocksRequestRepetition, t)
+ test.CheckInt("BlocksRequestMaxIdleRounds", c.BlocksRequestMaxIdleRounds, blocksRequestMaxIdleRounds, t)
+ test.CheckDuration("BlockHashesRequestInterval", c.BlockHashesRequestInterval, 300*time.Millisecond, t)
+ test.CheckDuration("BlocksRequestInterval", c.BlocksRequestInterval, 100*time.Millisecond, t)
+ test.CheckDuration("BlockHashesTimeout", c.BlockHashesTimeout, 90*time.Second, t)
+ test.CheckDuration("BlocksTimeout", c.BlocksTimeout, blocksTimeout, t)
+}
diff --git a/blockpool/errors_test.go b/blockpool/errors_test.go
new file mode 100644
index 000000000..65a161233
--- /dev/null
+++ b/blockpool/errors_test.go
@@ -0,0 +1,124 @@
+package blockpool
+
+import (
+ "testing"
+ "time"
+
+ "github.com/ethereum/go-ethereum/blockpool/test"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/pow"
+)
+
+func TestInvalidBlock(t *testing.T) {
+ test.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.serveBlocks(2, 3)
+ go peer1.serveBlockHashes(3, 2, 1, 0)
+ peer1.serveBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout)
+ 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 %v error, got %v", ErrInvalidBlock, peer1.peerErrors)
+ }
+}
+
+func TestVerifyPoW(t *testing.T) {
+ test.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)
+ peer2 := blockPoolTester.newPeer("peer2", 1, 3)
+ peer1.AddPeer()
+ peer2.AddPeer()
+ go peer1.serveBlocks(2, 3)
+ go peer1.serveBlockHashes(3, 2, 1, 0)
+ peer1.serveBlocks(0, 1, 2, 3)
+ blockPoolTester.blockPool.verifyPoW = func(b pow.Block) bool {
+ return true
+ }
+ peer2.serveBlocks(1, 2)
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[3] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+ if len(peer1.peerErrors) == 1 {
+ if peer1.peerErrors[0] != ErrInvalidPoW {
+ t.Errorf("wrong error, expected %v, got %v", ErrInvalidPoW, peer1.peerErrors[0])
+ }
+ } else {
+ t.Errorf("expected %v error, got %v", ErrInvalidPoW, peer1.peerErrors)
+ }
+}
+
+func TestUnrequestedBlock(t *testing.T) {
+ test.LogInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 3)
+ peer1.AddPeer()
+ peer1.sendBlocks(1, 2)
+
+ // blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ if len(peer1.peerErrors) == 1 {
+ if peer1.peerErrors[0] != ErrUnrequestedBlock {
+ t.Errorf("wrong error, got %v, expected %v", peer1.peerErrors[0], ErrUnrequestedBlock)
+ }
+ } else {
+ t.Errorf("expected %v error, got %v", ErrUnrequestedBlock, peer1.peerErrors)
+ }
+}
+
+func TestErrInsufficientChainInfo(t *testing.T) {
+ test.LogInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPool.Config.BlockHashesTimeout = 100 * time.Millisecond
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 3)
+ peer1.AddPeer()
+
+ blockPool.Wait(waitTimeout)
+ blockPool.Stop()
+ if len(peer1.peerErrors) == 1 {
+ if peer1.peerErrors[0] != ErrInsufficientChainInfo {
+ t.Errorf("wrong error, got %v, expected %v", peer1.peerErrors[0], ErrInsufficientChainInfo)
+ }
+ } else {
+ t.Errorf("expected %v error, got %v", ErrInsufficientChainInfo, peer1.peerErrors)
+ }
+}
diff --git a/blockpool/peers.go b/blockpool/peers.go
new file mode 100644
index 000000000..5f1b2017c
--- /dev/null
+++ b/blockpool/peers.go
@@ -0,0 +1,536 @@
+package blockpool
+
+import (
+ "bytes"
+ "math/big"
+ "math/rand"
+ "sort"
+ "sync"
+ "time"
+
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/errs"
+ "github.com/ethereum/go-ethereum/ethutil"
+)
+
+type peer struct {
+ lock sync.RWMutex
+
+ // last known blockchain status
+ td *big.Int
+ currentBlockHash []byte
+ currentBlock *types.Block
+ parentHash []byte
+ headSection *section
+
+ id string
+
+ // peer callbacks
+ requestBlockHashes func([]byte) error
+ requestBlocks func([][]byte) error
+ peerError func(*errs.Error)
+ errors *errs.Errors
+
+ sections [][]byte
+
+ // channels to push new head block and head section for peer a
+ currentBlockC chan *types.Block
+ headSectionC chan *section
+
+ // channels to signal peers witch and peer quit
+ idleC chan bool
+ switchC chan bool
+
+ quit chan bool
+ bp *BlockPool
+
+ // timers for head section process
+ blockHashesRequestTimer <-chan time.Time
+ blocksRequestTimer <-chan time.Time
+ suicide <-chan time.Time
+
+ idle bool
+}
+
+// peers is the component keeping a record of peers in a hashmap
+//
+type peers struct {
+ lock sync.RWMutex
+
+ bp *BlockPool
+ errors *errs.Errors
+ peers map[string]*peer
+ best *peer
+ status *status
+}
+
+// peer constructor
+func (self *peers) newPeer(
+ td *big.Int,
+ currentBlockHash []byte,
+ id string,
+ requestBlockHashes func([]byte) error,
+ requestBlocks func([][]byte) error,
+ peerError func(*errs.Error),
+) (p *peer) {
+
+ p = &peer{
+ errors: self.errors,
+ td: td,
+ currentBlockHash: currentBlockHash,
+ id: id,
+ requestBlockHashes: requestBlockHashes,
+ requestBlocks: requestBlocks,
+ peerError: peerError,
+ currentBlockC: make(chan *types.Block),
+ headSectionC: make(chan *section),
+ bp: self.bp,
+ idle: true,
+ }
+ // at creation the peer is recorded in the peer pool
+ self.peers[id] = p
+ return
+}
+
+// dispatches an error to a peer if still connected
+func (self *peers) peerError(id string, code int, format string, params ...interface{}) {
+ self.lock.RLock()
+ defer self.lock.RUnlock()
+ peer, ok := self.peers[id]
+ if ok {
+ peer.addError(code, format, params)
+ }
+ // blacklisting comes here
+}
+
+func (self *peer) addError(code int, format string, params ...interface{}) {
+ err := self.errors.New(code, format, params...)
+ self.peerError(err)
+}
+
+func (self *peer) setChainInfo(td *big.Int, c []byte) {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+
+ self.td = td
+ self.currentBlockHash = c
+
+ self.currentBlock = nil
+ self.parentHash = nil
+ self.headSection = nil
+}
+
+func (self *peer) setChainInfoFromBlock(block *types.Block) {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ // use the optional TD to update peer td, this helps second best peer selection
+ // in case best peer is lost
+ if block.Td != nil && block.Td.Cmp(self.td) > 0 {
+ plog.DebugDetailf("setChainInfoFromBlock: update <%s> - head: %v->%v - TD: %v->%v", self.id, hex(self.currentBlockHash), hex(block.Hash()), self.td, block.Td)
+ self.td = block.Td
+ self.currentBlockHash = block.Hash()
+ self.parentHash = block.ParentHash()
+ self.currentBlock = block
+ self.headSection = nil
+ }
+ self.bp.wg.Add(1)
+ go func() {
+ self.currentBlockC <- block
+ self.bp.wg.Done()
+ }()
+}
+
+func (self *peers) requestBlocks(attempts int, hashes [][]byte) {
+ // distribute block request among known peers
+ self.lock.RLock()
+ defer self.lock.RUnlock()
+ peerCount := len(self.peers)
+ // on first attempt use the best peer
+ if attempts == 0 {
+ plog.DebugDetailf("request %v missing blocks from best peer <%s>", len(hashes), self.best.id)
+ self.best.requestBlocks(hashes)
+ return
+ }
+ repetitions := self.bp.Config.BlocksRequestRepetition
+ if repetitions > peerCount {
+ repetitions = peerCount
+ }
+ i := 0
+ indexes := rand.Perm(peerCount)[0:repetitions]
+ sort.Ints(indexes)
+
+ plog.DebugDetailf("request %v missing blocks from %v/%v peers", len(hashes), repetitions, peerCount)
+ for _, peer := range self.peers {
+ if i == indexes[0] {
+ plog.DebugDetailf("request length: %v", len(hashes))
+ plog.DebugDetailf("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.bp.putHashSlice(hashes)
+}
+
+// addPeer implements the logic for blockpool.AddPeer
+// returns true iff peer is promoted as best peer in the pool
+func (self *peers) addPeer(
+ td *big.Int,
+ currentBlockHash []byte,
+ id string,
+ requestBlockHashes func([]byte) error,
+ requestBlocks func([][]byte) error,
+ peerError func(*errs.Error),
+) (best bool) {
+
+ var previousBlockHash []byte
+ self.lock.Lock()
+ p, found := self.peers[id]
+ if found {
+ if !bytes.Equal(p.currentBlockHash, currentBlockHash) {
+ previousBlockHash = p.currentBlockHash
+ plog.Debugf("addPeer: Update peer <%s> with td %v and current block %s (was %v)", id, td, hex(currentBlockHash), hex(previousBlockHash))
+ p.setChainInfo(td, currentBlockHash)
+ self.status.lock.Lock()
+ self.status.values.NewBlocks++
+ self.status.lock.Unlock()
+ }
+ } else {
+ p = self.newPeer(td, currentBlockHash, id, requestBlockHashes, requestBlocks, peerError)
+
+ self.status.lock.Lock()
+
+ self.status.peers[id]++
+ self.status.values.NewBlocks++
+ self.status.lock.Unlock()
+
+ plog.Debugf("addPeer: add new peer <%v> with td %v and current block %s", id, td, hex(currentBlockHash))
+ }
+ self.lock.Unlock()
+
+ // check peer current head
+ if self.bp.hasBlock(currentBlockHash) {
+ // peer not ahead
+ return false
+ }
+
+ if self.best == p {
+ // new block update for active current best peer -> request hashes
+ plog.Debugf("addPeer: <%s> already the best peer. Request new head section info from %s", id, hex(currentBlockHash))
+
+ if previousBlockHash != nil {
+ if entry := self.bp.get(previousBlockHash); entry != nil {
+ p.headSectionC <- nil
+ self.bp.activateChain(entry.section, p, nil)
+ p.sections = append(p.sections, previousBlockHash)
+ }
+ }
+ best = true
+ } else {
+ currentTD := ethutil.Big0
+ if self.best != nil {
+ currentTD = self.best.td
+ }
+ if td.Cmp(currentTD) > 0 {
+ self.status.lock.Lock()
+ self.status.bestPeers[p.id]++
+ self.status.lock.Unlock()
+ plog.Debugf("addPeer: peer <%v> promoted best peer", id)
+ self.bp.switchPeer(self.best, p)
+ self.best = p
+ best = true
+ }
+ }
+ return
+}
+
+// removePeer is called (via RemovePeer) by the eth protocol when the peer disconnects
+func (self *peers) removePeer(id string) {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+
+ p, found := self.peers[id]
+ if !found {
+ return
+ }
+
+ delete(self.peers, id)
+ plog.Debugf("addPeer: remove peer <%v>", id)
+
+ // if current best peer is removed, need to find a better one
+ if self.best == p {
+ var newp *peer
+ // FIXME: own TD
+ max := ethutil.Big0
+ // peer with the highest self-acclaimed TD is chosen
+ for _, pp := range self.peers {
+ if pp.td.Cmp(max) > 0 {
+ max = pp.td
+ newp = pp
+ }
+ }
+ if newp != nil {
+ self.status.lock.Lock()
+ self.status.bestPeers[p.id]++
+ self.status.lock.Unlock()
+ plog.Debugf("addPeer: peer <%v> with td %v promoted best peer", newp.id, newp.td)
+ } else {
+ plog.Warnln("addPeer: no suitable peers found")
+ }
+ self.best = newp
+ self.bp.switchPeer(p, newp)
+ }
+}
+
+// switchPeer launches section processes based on information about
+// shared interest and legacy of peers
+func (self *BlockPool) switchPeer(oldp, newp *peer) {
+
+ // first quit AddBlockHashes, requestHeadSection and activateChain
+ if oldp != nil {
+ plog.DebugDetailf("<%s> quit peer processes", oldp.id)
+ close(oldp.switchC)
+ }
+ if newp != nil {
+ newp.idleC = make(chan bool)
+ newp.switchC = make(chan bool)
+ // if new best peer has no head section yet, create it and run it
+ // otherwise head section is an element of peer.sections
+ if newp.headSection == nil {
+ plog.DebugDetailf("[%s] head section for [%s] not created, requesting info", newp.id, hex(newp.currentBlockHash))
+
+ if newp.idle {
+ self.wg.Add(1)
+ newp.idle = false
+ self.syncing()
+ }
+
+ go func() {
+ newp.run()
+ if !newp.idle {
+ self.wg.Done()
+ newp.idle = true
+ }
+ }()
+
+ }
+
+ var connected = make(map[string]*section)
+ var sections [][]byte
+ for _, hash := range newp.sections {
+ plog.DebugDetailf("activate chain starting from section [%s]", hex(hash))
+ // if section not connected (ie, top of a contiguous sequence of sections)
+ if connected[string(hash)] == nil {
+ // if not deleted, then reread from pool (it can be orphaned top half of a split section)
+ if entry := self.get(hash); entry != nil {
+ self.activateChain(entry.section, newp, connected)
+ connected[string(hash)] = entry.section
+ sections = append(sections, hash)
+ }
+ }
+ }
+ plog.DebugDetailf("<%s> section processes (%v non-contiguous sequences, was %v before)", newp.id, len(sections), len(newp.sections))
+ // need to lock now that newp is exposed to section processes
+ newp.lock.Lock()
+ newp.sections = sections
+ newp.lock.Unlock()
+ }
+ // finally deactivate section process for sections where newp didnt activate
+ // newp activating section process changes the quit channel for this reason
+ if oldp != nil {
+ plog.DebugDetailf("<%s> quit section processes", oldp.id)
+ //
+ close(oldp.idleC)
+ }
+}
+
+func (self *peers) getPeer(id string) (p *peer, best bool) {
+ self.lock.RLock()
+ defer self.lock.RUnlock()
+ if self.best != nil && self.best.id == id {
+ return self.best, true
+ }
+ p = self.peers[id]
+ return
+}
+
+func (self *peer) handleSection(sec *section) {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+
+ self.headSection = sec
+ self.blockHashesRequestTimer = nil
+
+ if sec == nil {
+ if self.idle {
+ self.idle = false
+ self.bp.wg.Add(1)
+ self.bp.syncing()
+ }
+
+ self.suicide = time.After(self.bp.Config.BlockHashesTimeout)
+
+ plog.DebugDetailf("HeadSection: <%s> head block hash changed (mined block received). New head %s", self.id, hex(self.currentBlockHash))
+ } else {
+ if !self.idle {
+ self.idle = true
+ self.suicide = nil
+ self.bp.wg.Done()
+ }
+ plog.DebugDetailf("HeadSection: <%s> head section [%s] created", self.id, sectionhex(sec))
+ }
+}
+
+func (self *peer) getCurrentBlock(currentBlock *types.Block) {
+ // called by update or after AddBlock signals that head block of current peer is received
+ if currentBlock == nil {
+ if entry := self.bp.get(self.currentBlockHash); entry != nil {
+ entry.node.lock.Lock()
+ currentBlock = entry.node.block
+ entry.node.lock.Unlock()
+ }
+ if currentBlock != nil {
+ plog.DebugDetailf("HeadSection: <%s> head block %s found in blockpool", self.id, hex(self.currentBlockHash))
+ } else {
+ plog.DebugDetailf("HeadSection: <%s> head block %s not found... requesting it", self.id, hex(self.currentBlockHash))
+ self.requestBlocks([][]byte{self.currentBlockHash})
+ self.blocksRequestTimer = time.After(self.bp.Config.BlocksRequestInterval)
+ return
+ }
+ } else {
+ plog.DebugDetailf("HeadSection: <%s> head block %s received (parent: %s)", self.id, hex(self.currentBlockHash), hex(currentBlock.ParentHash()))
+ }
+
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ self.currentBlock = currentBlock
+ self.parentHash = currentBlock.ParentHash()
+ plog.DebugDetailf("HeadSection: <%s> head block %s found (parent: [%s])... requesting hashes", self.id, hex(self.currentBlockHash), hex(self.parentHash))
+ self.blockHashesRequestTimer = time.After(0)
+ self.blocksRequestTimer = nil
+}
+
+func (self *peer) getBlockHashes() {
+ //if connecting parent is found
+ if self.bp.hasBlock(self.parentHash) {
+ plog.DebugDetailf("HeadSection: <%s> parent block %s found in blockchain", self.id, hex(self.parentHash))
+ err := self.bp.insertChain(types.Blocks([]*types.Block{self.currentBlock}))
+ if err != nil {
+ self.addError(ErrInvalidBlock, "%v", err)
+
+ self.bp.status.lock.Lock()
+ self.bp.status.badPeers[self.id]++
+ self.bp.status.lock.Unlock()
+ }
+ } else {
+ if parent := self.bp.get(self.parentHash); parent != nil {
+ if self.bp.get(self.currentBlockHash) == nil {
+ plog.DebugDetailf("HeadSection: <%s> connecting parent %s found in pool... creating singleton section", self.id, hex(self.parentHash))
+ n := &node{
+ hash: self.currentBlockHash,
+ block: self.currentBlock,
+ hashBy: self.id,
+ blockBy: self.id,
+ }
+ self.bp.newSection([]*node{n}).activate(self)
+ } else {
+ plog.DebugDetailf("HeadSection: <%s> connecting parent %s found in pool...head section [%s] exists...not requesting hashes", self.id, hex(self.parentHash), sectionhex(parent.section))
+ self.bp.activateChain(parent.section, self, nil)
+ }
+ } else {
+ plog.DebugDetailf("HeadSection: <%s> section [%s] requestBlockHashes", self.id, sectionhex(self.headSection))
+ self.requestBlockHashes(self.currentBlockHash)
+ self.blockHashesRequestTimer = time.After(self.bp.Config.BlockHashesRequestInterval)
+ return
+ }
+ }
+ self.blockHashesRequestTimer = nil
+ if !self.idle {
+ self.idle = true
+ self.suicide = nil
+ self.bp.wg.Done()
+ }
+}
+
+// main loop for head section process
+func (self *peer) run() {
+
+ self.lock.RLock()
+ switchC := self.switchC
+ currentBlockHash := self.currentBlockHash
+ self.lock.RUnlock()
+
+ self.blockHashesRequestTimer = nil
+
+ self.blocksRequestTimer = time.After(0)
+ self.suicide = time.After(self.bp.Config.BlockHashesTimeout)
+
+ var quit chan bool
+
+ var ping = time.NewTicker(5 * time.Second)
+
+LOOP:
+ for {
+ select {
+ case <-ping.C:
+ plog.Debugf("HeadSection: <%s> section with head %s, idle: %v", self.id, hex(self.currentBlockHash), self.idle)
+
+ // signal from AddBlockHashes that head section for current best peer is created
+ // if sec == nil, it signals that chain info has updated (new block message)
+ case sec := <-self.headSectionC:
+ self.handleSection(sec)
+ // local var quit channel is linked to sections suicide channel so that
+ if sec == nil {
+ quit = nil
+ } else {
+ quit = sec.suicideC
+ }
+
+ // periodic check for block hashes or parent block/section
+ case <-self.blockHashesRequestTimer:
+ self.getBlockHashes()
+
+ // signal from AddBlock that head block of current best peer has been received
+ case currentBlock := <-self.currentBlockC:
+ self.getCurrentBlock(currentBlock)
+
+ // keep requesting until found or timed out
+ case <-self.blocksRequestTimer:
+ self.getCurrentBlock(nil)
+
+ // quitting on timeout
+ case <-self.suicide:
+ self.peerError(self.bp.peers.errors.New(ErrInsufficientChainInfo, "timed out without providing block hashes or head block", currentBlockHash))
+
+ self.bp.status.lock.Lock()
+ self.bp.status.badPeers[self.id]++
+ self.bp.status.lock.Unlock()
+ // there is no persistence here, so GC will just take care of cleaning up
+ break LOOP
+
+ // signal for peer switch, quit
+ case <-switchC:
+ var complete = "incomplete "
+ if self.idle {
+ complete = "complete"
+ }
+ plog.Debugf("HeadSection: <%s> section with head %s %s... quit request loop due to peer switch", self.id, hex(self.currentBlockHash), complete)
+ break LOOP
+
+ // global quit for blockpool
+ case <-self.bp.quit:
+ break LOOP
+
+ // quit
+ case <-quit:
+ break LOOP
+ }
+ }
+ if !self.idle {
+ self.idle = true
+ self.bp.wg.Done()
+ }
+}
diff --git a/blockpool/peers_test.go b/blockpool/peers_test.go
new file mode 100644
index 000000000..e53d7160b
--- /dev/null
+++ b/blockpool/peers_test.go
@@ -0,0 +1,120 @@
+package blockpool
+
+import (
+ "math/big"
+ "testing"
+
+ "github.com/ethereum/go-ethereum/blockpool/test"
+)
+
+// the actual tests
+func TestAddPeer(t *testing.T) {
+ test.LogInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ peer0 := blockPoolTester.newPeer("peer0", 1, 0)
+ peer1 := blockPoolTester.newPeer("peer1", 2, 1)
+ peer2 := blockPoolTester.newPeer("peer2", 3, 2)
+ var bestpeer *peer
+
+ blockPool.Start()
+
+ // pool
+ best := peer0.AddPeer()
+ if !best {
+ t.Errorf("peer0 (TD=1) not accepted as best")
+ }
+ if blockPool.peers.best.id != "peer0" {
+ t.Errorf("peer0 (TD=1) not set as best")
+ }
+
+ best = peer2.AddPeer()
+ if !best {
+ t.Errorf("peer2 (TD=3) not accepted as best")
+ }
+ if blockPool.peers.best.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.peers.best.id != "peer2" {
+ t.Errorf("peer2 (TD=3) not set any more as best")
+ }
+ if blockPool.peers.best.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.peers.best.id != "peer2" {
+ t.Errorf("peer2 (TD=4) not set as best")
+ }
+ if blockPool.peers.best.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.peers.best.id == "peer1" {
+ t.Errorf("peer1 (TD=3) should not be set as best")
+ }
+ bestpeer, best = blockPool.peers.getPeer("peer1")
+ if bestpeer.td.Cmp(big.NewInt(int64(3))) != 0 {
+ t.Errorf("peer1 TD should be updated")
+ }
+
+ blockPool.RemovePeer("peer2")
+ bestpeer, best = blockPool.peers.getPeer("peer2")
+ if bestpeer != nil {
+ t.Errorf("peer2 not removed")
+ }
+
+ if blockPool.peers.best.id != "peer1" {
+ t.Errorf("existing peer1 (TD=3) should be set as best peer")
+ }
+ peer1.waitBlocksRequests(2)
+
+ blockPool.RemovePeer("peer1")
+ bestpeer, best = blockPool.peers.getPeer("peer1")
+ if bestpeer != nil {
+ t.Errorf("peer1 not removed")
+ }
+
+ if blockPool.peers.best.id != "peer0" {
+ t.Errorf("existing peer0 (TD=1) should be set as best peer")
+ }
+ peer0.waitBlocksRequests(0)
+
+ blockPool.RemovePeer("peer0")
+ bestpeer, best = blockPool.peers.getPeer("peer0")
+ if bestpeer != 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.peers.best.id != "peer0" {
+ t.Errorf("peer0 (TD=1) should be set as best")
+ }
+ peer0.waitBlocksRequests(3)
+
+ blockPool.Stop()
+
+}
diff --git a/blockpool/section.go b/blockpool/section.go
new file mode 100644
index 000000000..48ea15d31
--- /dev/null
+++ b/blockpool/section.go
@@ -0,0 +1,677 @@
+package blockpool
+
+import (
+ "sync"
+ "time"
+
+ "github.com/ethereum/go-ethereum/core/types"
+)
+
+/*
+ section is the worker on each chain section in the block pool
+ - remove the section if there are blocks missing after an absolute time
+ - remove 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 removed (inserted in blockchain, invalid or expired)
+ - 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
+*/
+type section struct {
+ lock sync.RWMutex
+
+ parent *section // connecting section back in time towards blockchain
+ child *section // connecting section forward in time
+
+ top *node // the topmost node = head node = youngest node within the chain section
+ bottom *node // the bottom node = root node = oldest node within the chain section
+ nodes []*node
+
+ peer *peer
+ parentHash []byte
+
+ blockHashes [][]byte
+
+ poolRootIndex int
+
+ bp *BlockPool
+
+ controlC chan *peer // to (de)register the current best peer
+ poolRootC chan *peer // indicate connectedness to blockchain (well, known blocks)
+ offC chan bool // closed if process terminated
+ suicideC chan bool // initiate suicide on the section
+ quitInitC chan bool // to signal end of initialisation
+ forkC chan chan bool // freeze section process while splitting
+ switchC chan bool // switching
+ idleC chan bool // channel to indicate thai food
+ processC chan *node //
+ missingC chan *node //
+
+ blocksRequestTimer <-chan time.Time
+ blockHashesRequestTimer <-chan time.Time
+ suicideTimer <-chan time.Time
+
+ blocksRequests int
+ blockHashesRequests int
+
+ blocksRequestsComplete bool
+ blockHashesRequestsComplete bool
+ ready bool
+ same bool
+ initialised bool
+ active bool
+
+ step int
+ idle int
+ missing int
+ lastMissing int
+ depth int
+ invalid bool
+ poolRoot bool
+}
+
+//
+func (self *BlockPool) newSection(nodes []*node) *section {
+ sec := &section{
+ bottom: nodes[len(nodes)-1],
+ top: nodes[0],
+ nodes: nodes,
+ poolRootIndex: len(nodes),
+ bp: self,
+ controlC: make(chan *peer),
+ poolRootC: make(chan *peer),
+ offC: make(chan bool),
+ }
+
+ for i, node := range nodes {
+ entry := &entry{node: node, section: sec, index: &index{i}}
+ self.set(node.hash, entry)
+ }
+
+ plog.DebugDetailf("[%s] setup section process", sectionhex(sec))
+
+ go sec.run()
+ return sec
+}
+
+func (self *section) addSectionToBlockChain(p *peer) {
+ self.bp.wg.Add(1)
+ go func() {
+
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ defer func() {
+ self.bp.wg.Done()
+ }()
+
+ var node *node
+ var keys []string
+ var blocks []*types.Block
+ for self.poolRootIndex > 0 {
+ node = self.nodes[self.poolRootIndex-1]
+ node.lock.RLock()
+ block := node.block
+ node.lock.RUnlock()
+ if block == nil {
+ break
+ }
+ self.poolRootIndex--
+ keys = append(keys, string(node.hash))
+ blocks = append(blocks, block)
+ }
+
+ if len(blocks) == 0 {
+ return
+ }
+
+ self.bp.lock.Lock()
+ for _, key := range keys {
+ delete(self.bp.pool, key)
+ }
+ self.bp.lock.Unlock()
+
+ plog.Infof("[%s] insert %v blocks [%v/%v] into blockchain", sectionhex(self), len(blocks), hex(blocks[0].Hash()), hex(blocks[len(blocks)-1].Hash()))
+ err := self.bp.insertChain(blocks)
+ if err != nil {
+ self.invalid = true
+ self.bp.peers.peerError(node.blockBy, ErrInvalidBlock, "%v", err)
+ plog.Warnf("invalid block %x", node.hash)
+ plog.Warnf("penalise peers %v (hash), %v (block)", node.hashBy, node.blockBy)
+
+ // or invalid block and the entire chain needs to be removed
+ self.removeInvalidChain()
+ } else {
+ // if all blocks inserted in this section
+ // then need to try to insert blocks in child section
+ if self.poolRootIndex == 0 {
+ // if there is a child section, then recursively call itself:
+ // also if section process is not terminated,
+ // then signal blockchain connectivity with poolRootC
+ if child := self.bp.getChild(self); child != nil {
+ select {
+ case <-child.offC:
+ plog.DebugDetailf("[%s] add complete child section [%s] to the blockchain", sectionhex(self), sectionhex(child))
+ case child.poolRootC <- p:
+ plog.DebugDetailf("[%s] add incomplete child section [%s] to the blockchain", sectionhex(self), sectionhex(child))
+ }
+ child.addSectionToBlockChain(p)
+ } else {
+ plog.DebugDetailf("[%s] no child section in pool", sectionhex(self))
+ }
+ plog.DebugDetailf("[%s] section completely inserted to blockchain - remove", sectionhex(self))
+ // complete sections are removed. if called from within section process,
+ // this must run in its own go routine to avoid deadlock
+ self.remove()
+ }
+ }
+
+ self.bp.status.lock.Lock()
+ if err == nil {
+ headKey := string(blocks[0].ParentHash())
+ height := self.bp.status.chain[headKey] + len(blocks)
+ self.bp.status.chain[string(blocks[len(blocks)-1].Hash())] = height
+ if height > self.bp.status.values.LongestChain {
+ self.bp.status.values.LongestChain = height
+ }
+ delete(self.bp.status.chain, headKey)
+ }
+ self.bp.status.values.BlocksInChain += len(blocks)
+ self.bp.status.values.BlocksInPool -= len(blocks)
+ if err != nil {
+ self.bp.status.badPeers[node.blockBy]++
+ }
+ self.bp.status.lock.Unlock()
+
+ }()
+
+}
+
+func (self *section) run() {
+
+ // absolute time after which sub-chain is killed if not complete (some blocks are missing)
+ self.suicideC = make(chan bool)
+ self.forkC = make(chan chan bool)
+ self.suicideTimer = time.After(self.bp.Config.BlocksTimeout)
+
+ // node channels for the section
+ // container for missing block hashes
+ var checking bool
+ var ping = time.NewTicker(5 * time.Second)
+
+LOOP:
+ for !self.blockHashesRequestsComplete || !self.blocksRequestsComplete {
+
+ select {
+ case <-ping.C:
+ var name = "no peer"
+ if self.peer != nil {
+ name = self.peer.id
+ }
+ plog.DebugDetailf("[%s] peer <%s> active: %v", sectionhex(self), name, self.active)
+
+ // global quit from blockpool
+ case <-self.bp.quit:
+ break LOOP
+
+ // pause for peer switching
+ case <-self.switchC:
+ self.switchC = nil
+
+ case p := <-self.poolRootC:
+ // signal on pool root channel indicates that the blockpool is
+ // connected to the blockchain, insert the longest chain of blocks
+ // ignored in idle mode to avoid inserting chain sections of non-live peers
+ self.poolRoot = true
+ // switch off hash requests in case they were on
+ self.blockHashesRequestTimer = nil
+ self.blockHashesRequestsComplete = true
+ self.switchOn(p)
+
+ // peer quit or demoted, put section in idle mode
+ case <-self.idleC:
+ // peer quit or demoted, put section in idle mode
+ plog.Debugf("[%s] peer <%s> quit or demoted", sectionhex(self), self.peer.id)
+ self.switchOff()
+ self.idleC = nil
+
+ // timebomb - if section is not complete in time, nuke the entire chain
+ case <-self.suicideTimer:
+ self.suicide()
+ plog.Debugf("[%s] timeout. (%v total attempts): missing %v/%v/%v...suicide", sectionhex(self), self.blocksRequests, self.missing, self.lastMissing, self.depth)
+ self.suicideTimer = nil
+
+ // closing suicideC triggers section suicide: removes section nodes from pool and terminates section process
+ case <-self.suicideC:
+ plog.DebugDetailf("[%s] suicide", sectionhex(self))
+ self.unlink()
+ self.bp.remove(self)
+ plog.DebugDetailf("[%s] done", sectionhex(self))
+ break LOOP
+
+ // alarm for checking blocks in the section
+ case <-self.blocksRequestTimer:
+ plog.DebugDetailf("[%s] alarm: block request time", sectionhex(self))
+ self.processC = self.missingC
+
+ // alarm for checking parent of the section or sending out hash requests
+ case <-self.blockHashesRequestTimer:
+ plog.DebugDetailf("[%s] alarm: hash request time", sectionhex(self))
+ self.blockHashesRequest()
+
+ // activate this section process with a peer
+ case p := <-self.controlC:
+ if p == nil {
+ self.switchOff()
+ } else {
+ self.switchOn(p)
+ }
+ self.bp.wg.Done()
+ // blocks the process until section is split at the fork
+ case waiter := <-self.forkC:
+ <-waiter
+ self.initialised = false
+ self.quitInitC = nil
+
+ //
+ case n, ok := <-self.processC:
+ // channel closed, first iteration finished
+ if !ok && !self.initialised {
+ plog.DebugDetailf("[%s] section initalised: missing %v/%v/%v", sectionhex(self), self.missing, self.lastMissing, self.depth)
+ self.initialised = true
+ self.processC = nil
+ // self.processC = make(chan *node, self.missing)
+ self.checkRound()
+ checking = false
+ break
+ }
+ plog.DebugDetailf("[%s] section proc step %v: missing %v/%v/%v", sectionhex(self), self.step, self.missing, self.lastMissing, self.depth)
+ if !checking {
+ self.step = 0
+ self.missing = 0
+ checking = true
+ }
+ self.step++
+
+ n.lock.RLock()
+ block := n.block
+ n.lock.RUnlock()
+
+ // if node has no block, request it (buffer it for batch request)
+ // feed it to missingC channel for the next round
+ if block == nil {
+ pos := self.missing % self.bp.Config.BlockBatchSize
+ if pos == 0 {
+ if self.missing != 0 {
+ self.bp.requestBlocks(self.blocksRequests, self.blockHashes[:])
+ }
+ self.blockHashes = self.bp.getHashSlice()
+ }
+ self.blockHashes[pos] = n.hash
+ self.missing++
+ self.missingC <- n
+ } else {
+ // checking for parent block
+ if self.poolRoot {
+ // if node has got block (received via async AddBlock call from protocol)
+ if self.step == self.lastMissing {
+ // current root of the pool
+ plog.DebugDetailf("[%s] received block for current pool root %s", sectionhex(self), hex(n.hash))
+ self.addSectionToBlockChain(self.peer)
+ }
+ } else {
+ if self.parentHash == nil && n == self.bottom {
+ self.parentHash = block.ParentHash()
+ plog.DebugDetailf("[%s] got parent head block hash %s...checking", sectionhex(self), hex(self.parentHash))
+ self.blockHashesRequest()
+ }
+ }
+ }
+ if self.initialised && self.step == self.lastMissing {
+ plog.DebugDetailf("[%s] check if new blocks arrived (attempt %v): missing %v/%v/%v", sectionhex(self), self.blocksRequests, self.missing, self.lastMissing, self.depth)
+ self.checkRound()
+ checking = false
+ }
+ } // select
+ } // for
+
+ close(self.offC)
+ if self.peer != nil {
+ self.active = false
+ self.bp.wg.Done()
+ }
+
+ plog.DebugDetailf("[%s] section process terminated: %v blocks retrieved (%v attempts), hash requests complete on root (%v attempts).", sectionhex(self), self.depth, self.blocksRequests, self.blockHashesRequests)
+
+}
+
+func (self *section) switchOn(newpeer *peer) {
+
+ oldpeer := self.peer
+ // reset switchC/switchC to current best peer
+ self.idleC = newpeer.idleC
+ self.switchC = newpeer.switchC
+ self.peer = newpeer
+
+ if oldpeer != newpeer {
+ oldp := "no peer"
+ newp := "no peer"
+ if oldpeer != nil {
+ oldp = oldpeer.id
+ }
+ if newpeer != nil {
+ newp = newpeer.id
+ }
+
+ plog.DebugDetailf("[%s] active mode <%s> -> <%s>", sectionhex(self), oldp, newp)
+ }
+
+ // activate section with current peer
+ if oldpeer == nil {
+ self.bp.wg.Add(1)
+ self.active = true
+
+ if !self.blockHashesRequestsComplete {
+ self.blockHashesRequestTimer = time.After(0)
+ }
+ if !self.blocksRequestsComplete {
+ if !self.initialised {
+ if self.quitInitC != nil {
+ <-self.quitInitC
+ }
+ self.missingC = make(chan *node, self.bp.Config.BlockHashesBatchSize)
+ self.processC = make(chan *node, self.bp.Config.BlockHashesBatchSize)
+ self.quitInitC = make(chan bool)
+
+ self.step = 0
+ self.missing = 0
+ self.depth = len(self.nodes)
+ self.lastMissing = self.depth
+
+ self.feedNodes()
+ } else {
+ self.blocksRequestTimer = time.After(0)
+ }
+ }
+ }
+}
+
+// put the section to idle mode
+func (self *section) switchOff() {
+ // active -> idle
+ if self.peer != nil {
+ oldp := "no peer"
+ oldpeer := self.peer
+ if oldpeer != nil {
+ oldp = oldpeer.id
+ }
+ plog.DebugDetailf("[%s] idle mode peer <%s> -> <> (%v total attempts): missing %v/%v/%v", sectionhex(self), oldp, self.blocksRequests, self.missing, self.lastMissing, self.depth)
+
+ self.active = false
+ self.peer = nil
+ // turn off timers
+ self.blocksRequestTimer = nil
+ self.blockHashesRequestTimer = nil
+
+ if self.quitInitC != nil {
+ <-self.quitInitC
+ self.quitInitC = nil
+ }
+ self.processC = nil
+ self.bp.wg.Done()
+ }
+}
+
+// iterates through nodes of a section to feed processC
+// used to initialise chain section
+func (self *section) feedNodes() {
+ // if not run at least once fully, launch iterator
+ self.bp.wg.Add(1)
+ go func() {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ defer func() {
+ self.bp.wg.Done()
+ }()
+ var n *node
+ INIT:
+ for _, n = range self.nodes {
+ select {
+ case self.processC <- n:
+ case <-self.bp.quit:
+ break INIT
+ }
+ }
+ close(self.processC)
+ close(self.quitInitC)
+ }()
+}
+
+func (self *section) blockHashesRequest() {
+
+ if self.switchC != nil {
+ self.bp.chainLock.Lock()
+ parentSection := self.parent
+
+ if parentSection == nil {
+
+ // only link to new parent if not switching peers
+ // this protects against synchronisation issue where during switching
+ // a demoted peer's fork will be chosen over the best peer's chain
+ // because relinking the correct chain (activateChain) is overwritten here in
+ // demoted peer's section process just before the section is put to idle mode
+ if self.parentHash != nil {
+ if parent := self.bp.get(self.parentHash); parent != nil {
+ parentSection = parent.section
+ plog.DebugDetailf("[%s] blockHashesRequest: parent section [%s] linked\n", sectionhex(self), sectionhex(parentSection))
+ link(parentSection, self)
+ } else {
+ if self.bp.hasBlock(self.parentHash) {
+ self.poolRoot = true
+ plog.DebugDetailf("[%s] blockHashesRequest: parentHash known ... inserting section in blockchain", sectionhex(self))
+ self.addSectionToBlockChain(self.peer)
+ self.blockHashesRequestTimer = nil
+ self.blockHashesRequestsComplete = true
+ }
+ }
+ }
+ }
+ self.bp.chainLock.Unlock()
+
+ if !self.poolRoot {
+ if parentSection != nil {
+ // activate parent section with this peer
+ // but only if not during switch mode
+ plog.DebugDetailf("[%s] parent section [%s] activated\n", sectionhex(self), sectionhex(parentSection))
+ self.bp.activateChain(parentSection, self.peer, nil)
+ // if not root of chain, switch off
+ plog.DebugDetailf("[%s] parent found, hash requests deactivated (after %v total attempts)\n", sectionhex(self), self.blockHashesRequests)
+ self.blockHashesRequestTimer = nil
+ self.blockHashesRequestsComplete = true
+ } else {
+ self.blockHashesRequests++
+ plog.DebugDetailf("[%s] hash request on root (%v total attempts)\n", sectionhex(self), self.blockHashesRequests)
+ self.peer.requestBlockHashes(self.bottom.hash)
+ self.blockHashesRequestTimer = time.After(self.bp.Config.BlockHashesRequestInterval)
+ }
+ }
+ }
+}
+
+// checks number of missing blocks after each round of request and acts accordingly
+func (self *section) checkRound() {
+ if self.missing == 0 {
+ // no missing blocks
+ plog.DebugDetailf("[%s] section checked: got all blocks. process complete (%v total blocksRequests): missing %v/%v/%v", sectionhex(self), self.blocksRequests, self.missing, self.lastMissing, self.depth)
+ self.blocksRequestsComplete = true
+ self.blocksRequestTimer = nil
+ } else {
+ // some missing blocks
+ plog.DebugDetailf("[%s] section checked: missing %v/%v/%v", sectionhex(self), self.missing, self.lastMissing, self.depth)
+ self.blocksRequests++
+ pos := self.missing % self.bp.Config.BlockBatchSize
+ if pos == 0 {
+ pos = self.bp.Config.BlockBatchSize
+ }
+ self.bp.requestBlocks(self.blocksRequests, self.blockHashes[:pos])
+
+ // handle idle rounds
+ if self.missing == self.lastMissing {
+ // idle round
+ if self.same {
+ // more than once
+ self.idle++
+ // too many idle rounds
+ if self.idle >= self.bp.Config.BlocksRequestMaxIdleRounds {
+ plog.DebugDetailf("[%s] block requests had %v idle rounds (%v total attempts): missing %v/%v/%v\ngiving up...", sectionhex(self), self.idle, self.blocksRequests, self.missing, self.lastMissing, self.depth)
+ self.suicide()
+ }
+ } else {
+ self.idle = 0
+ }
+ self.same = true
+ } else {
+ self.same = false
+ }
+ self.lastMissing = self.missing
+ // put processC offline
+ self.processC = nil
+ self.blocksRequestTimer = time.After(self.bp.Config.BlocksRequestInterval)
+ }
+}
+
+/*
+ link connects two sections via parent/child fields
+ creating a doubly linked list
+ caller must hold BlockPool chainLock
+*/
+func link(parent *section, child *section) {
+ if parent != nil {
+ exChild := parent.child
+ parent.child = child
+ if exChild != nil && exChild != child {
+ if child != nil {
+ // if child is nil it is not a real fork
+ plog.DebugDetailf("[%s] chain fork [%s] -> [%s]", sectionhex(parent), sectionhex(exChild), sectionhex(child))
+ }
+ exChild.parent = nil
+ }
+ }
+ if child != nil {
+ exParent := child.parent
+ if exParent != nil && exParent != parent {
+ if parent != nil {
+ // if parent is nil it is not a real fork, but suicide delinking section
+ plog.DebugDetailf("[%s] chain reverse fork [%s] -> [%s]", sectionhex(child), sectionhex(exParent), sectionhex(parent))
+ }
+ exParent.child = nil
+ }
+ child.parent = parent
+ }
+}
+
+/*
+ handle forks where connecting node is mid-section
+ by splitting section at fork
+ no splitting needed if connecting node is head of a section
+ caller must hold chain lock
+*/
+func (self *BlockPool) splitSection(parent *section, entry *entry) {
+ plog.DebugDetailf("[%s] split section at fork", sectionhex(parent))
+ parent.deactivate()
+ waiter := make(chan bool)
+ parent.wait(waiter)
+ chain := parent.nodes
+ parent.nodes = chain[entry.index.int:]
+ parent.top = parent.nodes[0]
+ parent.poolRootIndex -= entry.index.int
+ orphan := self.newSection(chain[0:entry.index.int])
+ link(orphan, parent.child)
+ close(waiter)
+ orphan.deactivate()
+}
+
+func (self *section) wait(waiter chan bool) {
+ self.forkC <- waiter
+}
+
+func (self *BlockPool) linkSections(nodes []*node, parent, child *section) (sec *section) {
+ // if new section is created, link it to parent/child sections
+ // and launch section process fetching block and further hashes
+ if len(nodes) > 0 {
+ sec = self.newSection(nodes)
+ plog.Debugf("[%s]->[%s](%v)->[%s] new chain section", sectionhex(parent), sectionhex(sec), len(nodes), sectionhex(child))
+ link(parent, sec)
+ link(sec, child)
+ } else {
+ // now this can only happen if we allow response to hash request to include <from> hash
+ // in this case we just link parent and child (without needing root block of child section)
+ plog.Debugf("[%s]->[%s] connecting known sections", sectionhex(parent), sectionhex(child))
+ link(parent, child)
+ }
+ return
+}
+
+func (self *section) activate(p *peer) {
+ self.bp.wg.Add(1)
+ select {
+ case <-self.offC:
+ self.bp.wg.Done()
+ case self.controlC <- p:
+ plog.DebugDetailf("[%s] activate section process for peer <%s>", sectionhex(self), p.id)
+ }
+}
+
+func (self *section) deactivate() {
+ self.bp.wg.Add(1)
+ self.controlC <- nil
+}
+
+func (self *section) suicide() {
+ select {
+ case <-self.suicideC:
+ return
+ default:
+ }
+ close(self.suicideC)
+}
+
+// removes this section exacly
+func (self *section) remove() {
+ select {
+ case <-self.offC:
+ // section is complete, no process
+ self.unlink()
+ self.bp.remove(self)
+ close(self.suicideC)
+ plog.DebugDetailf("[%s] remove: suicide", sectionhex(self))
+ case <-self.suicideC:
+ plog.DebugDetailf("[%s] remove: suicided already", sectionhex(self))
+ default:
+ plog.DebugDetailf("[%s] remove: suicide", sectionhex(self))
+ close(self.suicideC)
+ }
+ plog.DebugDetailf("[%s] removed section.", sectionhex(self))
+
+}
+
+// remove a section and all its descendents from the pool
+func (self *section) removeInvalidChain() {
+ // need to get the child before removeSection delinks the section
+ self.bp.chainLock.RLock()
+ child := self.child
+ self.bp.chainLock.RUnlock()
+
+ plog.DebugDetailf("[%s] remove invalid chain", sectionhex(self))
+ self.remove()
+ if child != nil {
+ child.removeInvalidChain()
+ }
+}
+
+// unlink a section from its parent/child
+func (self *section) unlink() {
+ // first delink from child and parent under chainlock
+ self.bp.chainLock.Lock()
+ link(nil, self)
+ link(self, nil)
+ self.bp.chainLock.Unlock()
+}
diff --git a/blockpool/status.go b/blockpool/status.go
new file mode 100644
index 000000000..0dd874232
--- /dev/null
+++ b/blockpool/status.go
@@ -0,0 +1,110 @@
+package blockpool
+
+import (
+ "fmt"
+ "sync"
+)
+
+type statusValues struct {
+ BlockHashes int // number of hashes fetched this session
+ BlockHashesInPool int // number of hashes currently in the pool
+ Blocks int // number of blocks fetched this session
+ BlocksInPool int // number of blocks currently in the pool
+ BlocksInChain int // number of blocks inserted/connected to the blockchain this session
+ NewBlocks int // number of new blocks (received with new blocks msg) this session
+ Forks int // number of chain forks in the blockchain (poolchain) this session
+ LongestChain int // the longest chain inserted since the start of session (aka session blockchain height)
+ BestPeer []byte //Pubkey
+ Syncing bool // requesting, updating etc
+ Peers int // cumulative number of all different registered peers since the start of this session
+ ActivePeers int // cumulative number of all different peers that contributed a hash or block since the start of this session
+ LivePeers int // number of live peers registered with the block pool (supposed to be redundant but good sanity check
+ BestPeers int // cumulative number of all peers that at some point were promoted as best peer (peer with highest TD status) this session
+ BadPeers int // cumulative number of all peers that violated the protocol (invalid block or pow, unrequested hash or block, etc)
+}
+
+type status struct {
+ lock sync.Mutex
+ values statusValues
+ chain map[string]int
+ peers map[string]int
+ bestPeers map[string]int
+ badPeers map[string]int
+ activePeers map[string]int
+}
+
+func newStatus() *status {
+ return &status{
+ chain: make(map[string]int),
+ peers: make(map[string]int),
+ bestPeers: make(map[string]int),
+ badPeers: make(map[string]int),
+ activePeers: make(map[string]int),
+ }
+}
+
+type Status struct {
+ statusValues
+}
+
+// blockpool status for reporting
+func (self *BlockPool) Status() *Status {
+ self.status.lock.Lock()
+ defer self.status.lock.Unlock()
+ self.status.values.BlockHashesInPool = len(self.pool)
+ self.status.values.ActivePeers = len(self.status.activePeers)
+ self.status.values.BestPeers = len(self.status.bestPeers)
+ self.status.values.BadPeers = len(self.status.badPeers)
+ self.status.values.LivePeers = len(self.peers.peers)
+ self.status.values.Peers = len(self.status.peers)
+ self.status.values.BlockHashesInPool = len(self.pool)
+ return &Status{self.status.values}
+}
+
+func (self *Status) String() string {
+ return fmt.Sprintf(`
+ Syncing: %v
+ BlockHashes: %v
+ BlockHashesInPool: %v
+ Blocks: %v
+ BlocksInPool: %v
+ BlocksInChain: %v
+ NewBlocks: %v
+ Forks: %v
+ LongestChain: %v
+ Peers: %v
+ LivePeers: %v
+ ActivePeers: %v
+ BestPeers: %v
+ BadPeers: %v
+`,
+ self.Syncing,
+ self.BlockHashes,
+ self.BlockHashesInPool,
+ self.Blocks,
+ self.BlocksInPool,
+ self.BlocksInChain,
+ self.NewBlocks,
+ self.Forks,
+ self.LongestChain,
+ self.Peers,
+ self.LivePeers,
+ self.ActivePeers,
+ self.BestPeers,
+ self.BadPeers,
+ )
+}
+
+func (self *BlockPool) syncing() {
+ self.status.lock.Lock()
+ defer self.status.lock.Unlock()
+ if !self.status.values.Syncing {
+ self.status.values.Syncing = true
+ go func() {
+ self.wg.Wait()
+ self.status.lock.Lock()
+ self.status.values.Syncing = false
+ self.status.lock.Unlock()
+ }()
+ }
+}
diff --git a/blockpool/status_test.go b/blockpool/status_test.go
new file mode 100644
index 000000000..7392f667a
--- /dev/null
+++ b/blockpool/status_test.go
@@ -0,0 +1,228 @@
+package blockpool
+
+import (
+ "fmt"
+ "testing"
+ // "time"
+
+ "github.com/ethereum/go-ethereum/blockpool/test"
+)
+
+var statusFields = []string{
+ "BlockHashes",
+ "BlockHashesInPool",
+ "Blocks",
+ "BlocksInPool",
+ "BlocksInChain",
+ "NewBlocks",
+ "Forks",
+ "LongestChain",
+ "Peers",
+ "LivePeers",
+ "ActivePeers",
+ "BestPeers",
+ "BadPeers",
+}
+
+func getStatusValues(s *Status) []int {
+ return []int{
+ s.BlockHashes,
+ s.BlockHashesInPool,
+ s.Blocks,
+ s.BlocksInPool,
+ s.BlocksInChain,
+ s.NewBlocks,
+ s.Forks,
+ s.LongestChain,
+ s.Peers,
+ s.LivePeers,
+ s.ActivePeers,
+ s.BestPeers,
+ s.BadPeers,
+ }
+}
+
+func checkStatus(t *testing.T, bp *BlockPool, syncing bool, expected []int) (err error) {
+ s := bp.Status()
+ if s.Syncing != syncing {
+ t.Errorf("status for Syncing incorrect. expected %v, got %v", syncing, s.Syncing)
+ }
+ got := getStatusValues(s)
+ for i, v := range expected {
+ err = test.CheckInt(statusFields[i], got[i], v, t)
+ if err != nil {
+ return err
+ }
+ fmt.Printf("%v: %v (%v)\n", statusFields[i], got[i], v)
+ }
+ return
+}
+
+// func TestBlockPoolStatus(t *testing.T) {
+// test.LogInit()
+// _, blockPool, blockPoolTester := newTestBlockPool(t)
+// blockPoolTester.blockChain[0] = nil
+// blockPoolTester.initRefBlockChain(12)
+// blockPoolTester.refBlockChain[3] = []int{4, 7}
+// delete(blockPoolTester.refBlockChain, 6)
+
+// blockPool.Start()
+
+// peer1 := blockPoolTester.newPeer("peer1", 1, 9)
+// peer2 := blockPoolTester.newPeer("peer2", 2, 6)
+// peer3 := blockPoolTester.newPeer("peer3", 3, 11)
+// peer4 := blockPoolTester.newPeer("peer4", 1, 9)
+// peer2.blocksRequestsMap = peer1.blocksRequestsMap
+
+// var expected []int
+// var err error
+// expected = []int{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}
+// err = checkStatus(t, blockPool, false, expected)
+// if err != nil {
+// return
+// }
+
+// peer1.AddPeer()
+// expected = []int{0, 0, 0, 0, 0, 1, 0, 0, 1, 1, 0, 1, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer1.serveBlocks(8, 9)
+// expected = []int{0, 0, 1, 1, 0, 1, 0, 0, 1, 1, 1, 1, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer1.serveBlockHashes(9, 8, 7, 3, 2)
+// expected = []int{5, 5, 1, 1, 0, 1, 0, 0, 1, 1, 1, 1, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer1.serveBlocks(3, 7, 8)
+// expected = []int{5, 5, 3, 3, 0, 1, 0, 0, 1, 1, 1, 1, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer1.serveBlocks(2, 3)
+// expected = []int{5, 5, 4, 4, 0, 1, 0, 0, 1, 1, 1, 1, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer4.AddPeer()
+// expected = []int{5, 5, 4, 4, 0, 2, 0, 0, 2, 2, 1, 1, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer4.sendBlockHashes(12, 11)
+// expected = []int{5, 5, 4, 4, 0, 2, 0, 0, 2, 2, 1, 1, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer2.AddPeer()
+// expected = []int{5, 5, 4, 4, 0, 3, 0, 0, 3, 3, 1, 2, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer2.serveBlocks(5, 6)
+// peer2.serveBlockHashes(6, 5, 4, 3, 2)
+// expected = []int{8, 8, 5, 5, 0, 3, 1, 0, 3, 3, 2, 2, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer2.serveBlocks(2, 3, 4)
+// expected = []int{8, 8, 6, 6, 0, 3, 1, 0, 3, 3, 2, 2, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// blockPool.RemovePeer("peer2")
+// expected = []int{8, 8, 6, 6, 0, 3, 1, 0, 3, 2, 2, 2, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer1.serveBlockHashes(2, 1, 0)
+// expected = []int{9, 9, 6, 6, 0, 3, 1, 0, 3, 2, 2, 2, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer1.serveBlocks(1, 2)
+// expected = []int{9, 9, 7, 7, 0, 3, 1, 0, 3, 2, 2, 2, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer1.serveBlocks(4, 5)
+// expected = []int{9, 9, 8, 8, 0, 3, 1, 0, 3, 2, 2, 2, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer3.AddPeer()
+// expected = []int{9, 9, 8, 8, 0, 4, 1, 0, 4, 3, 2, 3, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer3.serveBlocks(10, 11)
+// expected = []int{9, 9, 9, 9, 0, 4, 1, 0, 4, 3, 3, 3, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer3.serveBlockHashes(11, 10, 9)
+// expected = []int{11, 11, 9, 9, 0, 4, 1, 0, 4, 3, 3, 3, 0}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer4.sendBlocks(11, 12)
+// expected = []int{11, 11, 9, 9, 0, 4, 1, 0, 4, 3, 4, 3, 1}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+// peer3.serveBlocks(9, 10)
+// expected = []int{11, 11, 10, 10, 0, 4, 1, 0, 4, 3, 4, 3, 1}
+// err = checkStatus(t, blockPool, true, expected)
+// if err != nil {
+// return
+// }
+
+// peer3.serveBlocks(0, 1)
+// blockPool.Wait(waitTimeout)
+// time.Sleep(200 * time.Millisecond)
+// expected = []int{11, 3, 11, 3, 8, 4, 1, 8, 4, 3, 4, 3, 1}
+// err = checkStatus(t, blockPool, false, expected)
+// if err != nil {
+// return
+// }
+
+// blockPool.Stop()
+// }
diff --git a/blockpool/test/hash_pool.go b/blockpool/test/hash_pool.go
new file mode 100644
index 000000000..4e0332d7d
--- /dev/null
+++ b/blockpool/test/hash_pool.go
@@ -0,0 +1,57 @@
+package test
+
+import (
+ "sync"
+
+ "github.com/ethereum/go-ethereum/crypto"
+)
+
+// test helpers
+// TODO: move into common test helper package (see p2p/crypto etc.)
+
+func NewHashPool() *TestHashPool {
+ return &TestHashPool{intToHash: make(intToHash), hashToInt: make(hashToInt)}
+}
+
+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
+}
diff --git a/blockpool/test/logger.go b/blockpool/test/logger.go
new file mode 100644
index 000000000..8b776e0b5
--- /dev/null
+++ b/blockpool/test/logger.go
@@ -0,0 +1,78 @@
+package test
+
+import (
+ "log"
+ "os"
+ "sync"
+ "testing"
+
+ "github.com/ethereum/go-ethereum/logger"
+)
+
+var once sync.Once
+
+/* usage:
+func TestFunc(t *testing.T) {
+ test.LogInit()
+ // test
+}
+*/
+func LogInit() {
+ once.Do(func() {
+ var logsys = logger.NewStdLogSystem(os.Stdout, log.LstdFlags, logger.LogLevel(logger.DebugDetailLevel))
+ logger.AddLogSystem(logsys)
+ })
+}
+
+type testLogger struct{ t *testing.T }
+
+/* usage:
+func TestFunc(t *testing.T) {
+ defer test.Testlog.Detach()
+ // test
+}
+*/
+func Testlog(t *testing.T) testLogger {
+ logger.Reset()
+ l := testLogger{t}
+ logger.AddLogSystem(l)
+ return l
+}
+
+func (testLogger) GetLogLevel() logger.LogLevel { return logger.DebugLevel }
+func (testLogger) SetLogLevel(logger.LogLevel) {}
+
+func (l testLogger) LogPrint(level logger.LogLevel, msg string) {
+ l.t.Logf("%s", msg)
+}
+
+func (testLogger) Detach() {
+ logger.Flush()
+ logger.Reset()
+}
+
+type benchLogger struct{ b *testing.B }
+
+/* usage:
+func BenchmarkFunc(b *testing.B) {
+ defer test.Benchlog.Detach()
+ // test
+}
+*/
+func Benchlog(b *testing.B) benchLogger {
+ logger.Reset()
+ l := benchLogger{b}
+ logger.AddLogSystem(l)
+ return l
+}
+
+func (benchLogger) GetLogLevel() logger.LogLevel { return logger.Silence }
+
+func (benchLogger) SetLogLevel(logger.LogLevel) {}
+func (l benchLogger) LogPrint(level logger.LogLevel, msg string) {
+ l.b.Logf("%s", msg)
+}
+func (benchLogger) Detach() {
+ logger.Flush()
+ logger.Reset()
+}
diff --git a/blockpool/test/util.go b/blockpool/test/util.go
new file mode 100644
index 000000000..e183bf1d1
--- /dev/null
+++ b/blockpool/test/util.go
@@ -0,0 +1,35 @@
+package test
+
+import (
+ "fmt"
+ "testing"
+ "time"
+)
+
+func CheckInt(name string, got int, expected int, t *testing.T) (err error) {
+ if got != expected {
+ t.Errorf("status for %v incorrect. expected %v, got %v", name, expected, got)
+ err = fmt.Errorf("")
+ }
+ return
+}
+
+func CheckDuration(name string, got time.Duration, expected time.Duration, t *testing.T) (err error) {
+ if got != expected {
+ t.Errorf("status for %v incorrect. expected %v, got %v", name, expected, got)
+ err = fmt.Errorf("")
+ }
+ return
+}
+
+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
+}
diff --git a/errs/errors.go b/errs/errors.go
new file mode 100644
index 000000000..f1b0ebdbb
--- /dev/null
+++ b/errs/errors.go
@@ -0,0 +1,91 @@
+package errs
+
+import (
+ "fmt"
+
+ "github.com/ethereum/go-ethereum/logger"
+)
+
+/*
+Errors implements an error handler providing standardised errors for a package.
+Fields:
+
+ Errors:
+ a map from error codes to description
+
+ Package:
+ name of the package/component
+
+ Level:
+ a function mapping error code to logger.LogLevel (severity)
+ if not given, errors default to logger.InfoLevel
+*/
+type Errors struct {
+ Errors map[int]string
+ Package string
+ Level func(code int) logger.LogLevel
+}
+
+/*
+Error implements the standard go error interface.
+
+ errors.New(code, format, params ...interface{})
+
+Prints as:
+
+ [package] description: details
+
+where details is fmt.Sprintf(self.format, self.params...)
+*/
+type Error struct {
+ Code int
+ Name string
+ Package string
+ level logger.LogLevel
+ message string
+ format string
+ params []interface{}
+}
+
+func (self *Errors) New(code int, format string, params ...interface{}) *Error {
+ name, ok := self.Errors[code]
+ if !ok {
+ panic("invalid error code")
+ }
+ level := logger.InfoLevel
+ if self.Level != nil {
+ level = self.Level(code)
+ }
+ return &Error{
+ Code: code,
+ Name: name,
+ Package: self.Package,
+ level: level,
+ format: format,
+ params: params,
+ }
+}
+
+func (self Error) Error() (message string) {
+ if len(message) == 0 {
+ self.message = fmt.Sprintf("[%s] %s", self.Package, self.Name)
+ if self.format != "" {
+ self.message += ": " + fmt.Sprintf(self.format, self.params...)
+ }
+ }
+ return self.message
+}
+
+func (self Error) Log(log *logger.Logger) {
+ log.Sendln(self.level, self)
+}
+
+/*
+err.Fatal() is true if err's severity level is 0 or 1 (logger.ErrorLevel or logger.Silence)
+*/
+func (self *Error) Fatal() (fatal bool) {
+ if self.level < logger.WarnLevel {
+ fatal = true
+ }
+ return
+}
diff --git a/errs/errors_test.go b/errs/errors_test.go
new file mode 100644
index 000000000..09f70eef5
--- /dev/null
+++ b/errs/errors_test.go
@@ -0,0 +1,46 @@
+package errs
+
+import (
+ "fmt"
+ "testing"
+
+ "github.com/ethereum/go-ethereum/logger"
+)
+
+func testErrors() *Errors {
+ return &Errors{
+ Package: "TEST",
+ Errors: map[int]string{
+ 0: "zero",
+ 1: "one",
+ },
+ Level: func(i int) (l logger.LogLevel) {
+ if i == 0 {
+ l = logger.ErrorLevel
+ } else {
+ l = logger.WarnLevel
+ }
+ return
+ },
+ }
+}
+
+func TestErrorMessage(t *testing.T) {
+ err := testErrors().New(0, "zero detail %v", "available")
+ message := fmt.Sprintf("%v", err)
+ exp := "[TEST] zero: zero detail available"
+ if message != exp {
+ t.Errorf("error message incorrect. expected %v, got %v", exp, message)
+ }
+}
+
+func TestErrorSeverity(t *testing.T) {
+ err0 := testErrors().New(0, "zero detail")
+ if !err0.Fatal() {
+ t.Errorf("error should be fatal")
+ }
+ err1 := testErrors().New(1, "one detail")
+ if err1.Fatal() {
+ t.Errorf("error should not be fatal")
+ }
+}
diff --git a/eth/backend.go b/eth/backend.go
index 0e5d24429..f6cde5ccd 100644
--- a/eth/backend.go
+++ b/eth/backend.go
@@ -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 }
diff --git a/eth/block_pool.go b/eth/block_pool.go
deleted file mode 100644
index 13016c694..000000000
--- a/eth/block_pool.go
+++ /dev/null
@@ -1,1239 +0,0 @@
-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
-}
diff --git a/eth/block_pool_test.go b/eth/block_pool_test.go
deleted file mode 100644
index 331dbe504..000000000
--- a/eth/block_pool_test.go
+++ /dev/null
@@ -1,983 +0,0 @@
-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)
-
-}
diff --git a/eth/error.go b/eth/error.go
deleted file mode 100644
index 9c4a68481..000000000
--- a/eth/error.go
+++ /dev/null
@@ -1,72 +0,0 @@
-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
-}
diff --git a/eth/protocol.go b/eth/protocol.go
index 8221c1b29..ee2316836 100644
--- a/eth/protocol.go
+++ b/eth/protocol.go
@@ -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() {
diff --git a/eth/protocol_test.go b/eth/protocol_test.go
index a91806a1c..f499d033e 100644
--- a/eth/protocol_test.go
+++ b/eth/protocol_test.go
@@ -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)
diff --git a/eth/test/README.md b/eth/test/README.md
deleted file mode 100644
index 65728efa5..000000000
--- a/eth/test/README.md
+++ /dev/null
@@ -1,27 +0,0 @@
-= 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
-
diff --git a/eth/test/bootstrap.sh b/eth/test/bootstrap.sh
deleted file mode 100644
index 3da038be8..000000000
--- a/eth/test/bootstrap.sh
+++ /dev/null
@@ -1,9 +0,0 @@
-#!/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
diff --git a/eth/test/chains/00.chain b/eth/test/chains/00.chain
deleted file mode 100755
index ad3c05b24..000000000
--- a/eth/test/chains/00.chain
+++ /dev/null
Binary files differ
diff --git a/eth/test/chains/01.chain b/eth/test/chains/01.chain
deleted file mode 100755
index 56c9aef65..000000000
--- a/eth/test/chains/01.chain
+++ /dev/null
Binary files differ
diff --git a/eth/test/chains/02.chain b/eth/test/chains/02.chain
deleted file mode 100755
index 440c92d65..000000000
--- a/eth/test/chains/02.chain
+++ /dev/null
Binary files differ
diff --git a/eth/test/chains/03.chain b/eth/test/chains/03.chain
deleted file mode 100755
index 1cc7570ab..000000000
--- a/eth/test/chains/03.chain
+++ /dev/null
Binary files differ
diff --git a/eth/test/chains/04.chain b/eth/test/chains/04.chain
deleted file mode 100755
index d4e5b1aa8..000000000
--- a/eth/test/chains/04.chain
+++ /dev/null
Binary files differ
diff --git a/eth/test/mine.sh b/eth/test/mine.sh
deleted file mode 100644
index 0d95db1e4..000000000
--- a/eth/test/mine.sh
+++ /dev/null
@@ -1,20 +0,0 @@
-#!/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'
diff --git a/eth/test/run.sh b/eth/test/run.sh
deleted file mode 100644
index 5229af035..000000000
--- a/eth/test/run.sh
+++ /dev/null
@@ -1,53 +0,0 @@
-#!/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
diff --git a/eth/test/tests/00.chain b/eth/test/tests/00.chain
deleted file mode 120000
index 9655cb3df..000000000
--- a/eth/test/tests/00.chain
+++ /dev/null
@@ -1 +0,0 @@
-../chains/01.chain \ No newline at end of file
diff --git a/eth/test/tests/00.sh b/eth/test/tests/00.sh
deleted file mode 100644
index 9c5077164..000000000
--- a/eth/test/tests/00.sh
+++ /dev/null
@@ -1,13 +0,0 @@
-#!/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
-
diff --git a/eth/test/tests/01.chain b/eth/test/tests/01.chain
deleted file mode 120000
index ae65ccb37..000000000
--- a/eth/test/tests/01.chain
+++ /dev/null
@@ -1 +0,0 @@
-../chains/02.chain \ No newline at end of file
diff --git a/eth/test/tests/01.sh b/eth/test/tests/01.sh
deleted file mode 100644
index 1601e0dfb..000000000
--- a/eth/test/tests/01.sh
+++ /dev/null
@@ -1,18 +0,0 @@
-#!/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
-
diff --git a/eth/test/tests/02.chain b/eth/test/tests/02.chain
deleted file mode 120000
index 9655cb3df..000000000
--- a/eth/test/tests/02.chain
+++ /dev/null
@@ -1 +0,0 @@
-../chains/01.chain \ No newline at end of file
diff --git a/eth/test/tests/02.sh b/eth/test/tests/02.sh
deleted file mode 100644
index 619217ed8..000000000
--- a/eth/test/tests/02.sh
+++ /dev/null
@@ -1,19 +0,0 @@
-#!/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
-
diff --git a/eth/test/tests/03.chain b/eth/test/tests/03.chain
deleted file mode 120000
index b07c49a30..000000000
--- a/eth/test/tests/03.chain
+++ /dev/null
@@ -1 +0,0 @@
-../chains/12k.chain \ No newline at end of file
diff --git a/eth/test/tests/03.sh b/eth/test/tests/03.sh
deleted file mode 100644
index d7dba737f..000000000
--- a/eth/test/tests/03.sh
+++ /dev/null
@@ -1,14 +0,0 @@
-#!/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
-
diff --git a/eth/test/tests/04.sh b/eth/test/tests/04.sh
deleted file mode 100644
index d77c360ba..000000000
--- a/eth/test/tests/04.sh
+++ /dev/null
@@ -1,17 +0,0 @@
-#!/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'
diff --git a/eth/test/tests/05.sh b/eth/test/tests/05.sh
deleted file mode 100644
index 0a0f94df5..000000000
--- a/eth/test/tests/05.sh
+++ /dev/null
@@ -1,20 +0,0 @@
-#!/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
-
diff --git a/eth/test/tests/common.js b/eth/test/tests/common.js
deleted file mode 100644
index 206ebf145..000000000
--- a/eth/test/tests/common.js
+++ /dev/null
@@ -1,9 +0,0 @@
-function log(text) {
- console.log("[JS TEST SCRIPT] " + text);
-}
-
-function sleep(seconds) {
- var now = new Date().getTime();
- while(new Date().getTime() < now + seconds){}
-}
-
diff --git a/eth/test/tests/common.sh b/eth/test/tests/common.sh
deleted file mode 100644
index f66cfb43e..000000000
--- a/eth/test/tests/common.sh
+++ /dev/null
@@ -1,20 +0,0 @@
-#!/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
diff --git a/logger/loggers.go b/logger/loggers.go
index 147b2b85f..25263853a 100644
--- a/logger/loggers.go
+++ b/logger/loggers.go
@@ -42,74 +42,74 @@ func NewLogger(tag string) *Logger {
return &Logger{"[" + tag + "] "}
}
-func (logger *Logger) sendln(level LogLevel, v ...interface{}) {
+func (logger *Logger) Sendln(level LogLevel, v ...interface{}) {
logMessageC <- message{level, logger.tag + fmt.Sprintln(v...)}
}
-func (logger *Logger) sendf(level LogLevel, format string, v ...interface{}) {
+func (logger *Logger) Sendf(level LogLevel, format string, v ...interface{}) {
logMessageC <- message{level, logger.tag + fmt.Sprintf(format, v...)}
}
// Errorln writes a message with ErrorLevel.
func (logger *Logger) Errorln(v ...interface{}) {
- logger.sendln(ErrorLevel, v...)
+ logger.Sendln(ErrorLevel, v...)
}
// Warnln writes a message with WarnLevel.
func (logger *Logger) Warnln(v ...interface{}) {
- logger.sendln(WarnLevel, v...)
+ logger.Sendln(WarnLevel, v...)
}
// Infoln writes a message with InfoLevel.
func (logger *Logger) Infoln(v ...interface{}) {
- logger.sendln(InfoLevel, v...)
+ logger.Sendln(InfoLevel, v...)
}
// Debugln writes a message with DebugLevel.
func (logger *Logger) Debugln(v ...interface{}) {
- logger.sendln(DebugLevel, v...)
+ logger.Sendln(DebugLevel, v...)
}
// DebugDetailln writes a message with DebugDetailLevel.
func (logger *Logger) DebugDetailln(v ...interface{}) {
- logger.sendln(DebugDetailLevel, v...)
+ logger.Sendln(DebugDetailLevel, v...)
}
// Errorf writes a message with ErrorLevel.
func (logger *Logger) Errorf(format string, v ...interface{}) {
- logger.sendf(ErrorLevel, format, v...)
+ logger.Sendf(ErrorLevel, format, v...)
}
// Warnf writes a message with WarnLevel.
func (logger *Logger) Warnf(format string, v ...interface{}) {
- logger.sendf(WarnLevel, format, v...)
+ logger.Sendf(WarnLevel, format, v...)
}
// Infof writes a message with InfoLevel.
func (logger *Logger) Infof(format string, v ...interface{}) {
- logger.sendf(InfoLevel, format, v...)
+ logger.Sendf(InfoLevel, format, v...)
}
// Debugf writes a message with DebugLevel.
func (logger *Logger) Debugf(format string, v ...interface{}) {
- logger.sendf(DebugLevel, format, v...)
+ logger.Sendf(DebugLevel, format, v...)
}
// DebugDetailf writes a message with DebugDetailLevel.
func (logger *Logger) DebugDetailf(format string, v ...interface{}) {
- logger.sendf(DebugDetailLevel, format, v...)
+ logger.Sendf(DebugDetailLevel, format, v...)
}
// Fatalln writes a message with ErrorLevel and exits the program.
func (logger *Logger) Fatalln(v ...interface{}) {
- logger.sendln(ErrorLevel, v...)
+ logger.Sendln(ErrorLevel, v...)
Flush()
os.Exit(0)
}
// Fatalf writes a message with ErrorLevel and exits the program.
func (logger *Logger) Fatalf(format string, v ...interface{}) {
- logger.sendf(ErrorLevel, format, v...)
+ logger.Sendf(ErrorLevel, format, v...)
Flush()
os.Exit(0)
}