diff options
author | zelig <viktor.tron@gmail.com> | 2015-02-25 21:06:59 +0800 |
---|---|---|
committer | zelig <viktor.tron@gmail.com> | 2015-02-25 21:38:37 +0800 |
commit | 16ecda951b767800b4e09ad8e86e0866b05136be (patch) | |
tree | dfb4defb949713c03e05988de1abae5123733cf5 | |
parent | cdc86965cb38335b97d0bef337e0a53ed7c7f7e3 (diff) | |
download | go-tangerine-16ecda951b767800b4e09ad8e86e0866b05136be.tar go-tangerine-16ecda951b767800b4e09ad8e86e0866b05136be.tar.gz go-tangerine-16ecda951b767800b4e09ad8e86e0866b05136be.tar.bz2 go-tangerine-16ecda951b767800b4e09ad8e86e0866b05136be.tar.lz go-tangerine-16ecda951b767800b4e09ad8e86e0866b05136be.tar.xz go-tangerine-16ecda951b767800b4e09ad8e86e0866b05136be.tar.zst go-tangerine-16ecda951b767800b4e09ad8e86e0866b05136be.zip |
integrate blockpool into eth
- remove blockpool code
- remove blockpool integration test (kinda embarrassing)
- remove errors.go
-rw-r--r-- | eth/backend.go | 7 | ||||
-rw-r--r-- | eth/block_pool.go | 1239 | ||||
-rw-r--r-- | eth/block_pool_test.go | 983 | ||||
-rw-r--r-- | eth/error.go | 72 | ||||
-rw-r--r-- | eth/protocol.go | 69 | ||||
-rw-r--r-- | eth/protocol_test.go | 18 | ||||
-rw-r--r-- | eth/test/README.md | 27 | ||||
-rw-r--r-- | eth/test/bootstrap.sh | 9 | ||||
-rwxr-xr-x | eth/test/chains/00.chain | bin | 9726 -> 0 bytes | |||
-rwxr-xr-x | eth/test/chains/01.chain | bin | 13881 -> 0 bytes | |||
-rwxr-xr-x | eth/test/chains/02.chain | bin | 14989 -> 0 bytes | |||
-rwxr-xr-x | eth/test/chains/03.chain | bin | 18590 -> 0 bytes | |||
-rwxr-xr-x | eth/test/chains/04.chain | bin | 20529 -> 0 bytes | |||
-rw-r--r-- | eth/test/mine.sh | 20 | ||||
-rw-r--r-- | eth/test/run.sh | 53 | ||||
l--------- | eth/test/tests/00.chain | 1 | ||||
-rw-r--r-- | eth/test/tests/00.sh | 13 | ||||
l--------- | eth/test/tests/01.chain | 1 | ||||
-rw-r--r-- | eth/test/tests/01.sh | 18 | ||||
l--------- | eth/test/tests/02.chain | 1 | ||||
-rw-r--r-- | eth/test/tests/02.sh | 19 | ||||
l--------- | eth/test/tests/03.chain | 1 | ||||
-rw-r--r-- | eth/test/tests/03.sh | 14 | ||||
-rw-r--r-- | eth/test/tests/04.sh | 17 | ||||
-rw-r--r-- | eth/test/tests/05.sh | 20 | ||||
-rw-r--r-- | eth/test/tests/common.js | 9 | ||||
-rw-r--r-- | eth/test/tests/common.sh | 20 |
27 files changed, 61 insertions, 2570 deletions
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 = §ion{ - 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 Binary files differdeleted file mode 100755 index ad3c05b24..000000000 --- a/eth/test/chains/00.chain +++ /dev/null diff --git a/eth/test/chains/01.chain b/eth/test/chains/01.chain Binary files differdeleted file mode 100755 index 56c9aef65..000000000 --- a/eth/test/chains/01.chain +++ /dev/null diff --git a/eth/test/chains/02.chain b/eth/test/chains/02.chain Binary files differdeleted file mode 100755 index 440c92d65..000000000 --- a/eth/test/chains/02.chain +++ /dev/null diff --git a/eth/test/chains/03.chain b/eth/test/chains/03.chain Binary files differdeleted file mode 100755 index 1cc7570ab..000000000 --- a/eth/test/chains/03.chain +++ /dev/null diff --git a/eth/test/chains/04.chain b/eth/test/chains/04.chain Binary files differdeleted file mode 100755 index d4e5b1aa8..000000000 --- a/eth/test/chains/04.chain +++ /dev/null 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 |