diff --git a/eth/block_pool.go b/eth/block_pool.go index 7cfbc63f86..65d58ab022 100644 --- a/eth/block_pool.go +++ b/eth/block_pool.go @@ -1,6 +1,7 @@ package eth import ( + "fmt" "math" "math/big" "math/rand" @@ -10,46 +11,54 @@ import ( "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/logger" "github.com/ethereum/go-ethereum/pow" ) -var poolLogger = ethlogger.NewLogger("Blockpool") +var poolLogger = logger.NewLogger("Blockpool") const ( blockHashesBatchSize = 256 blockBatchSize = 64 - blocksRequestInterval = 10 // seconds + blocksRequestInterval = 500 // ms blocksRequestRepetition = 1 - blockHashesRequestInterval = 10 // seconds - blocksRequestMaxIdleRounds = 10 + blockHashesRequestInterval = 500 // ms + blocksRequestMaxIdleRounds = 100 cacheTimeout = 3 // minutes blockTimeout = 5 // minutes ) type poolNode struct { - lock sync.RWMutex - hash []byte - block *types.Block - child *poolNode - parent *poolNode - section *section - knownParent bool - peer string - source string - complete bool + 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 - pool map[string]*poolNode + 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 @@ -70,8 +79,23 @@ type peerInfo struct { peerError func(int, string, ...interface{}) sections map[string]*section - roots []*poolNode - quitC chan bool + + 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 bool + suicideC chan bool + blockChainC chan bool + forkC chan chan bool + off bool } func NewBlockPool(hasBlock func(hash []byte) bool, insertChain func(types.Blocks) error, verifyPoW func(pow.Block) bool, @@ -92,7 +116,9 @@ func (self *BlockPool) Start() { } self.running = true self.quit = make(chan bool) - self.pool = make(map[string]*poolNode) + self.flushC = make(chan bool) + self.pool = make(map[string]*poolEntry) + self.lock.Unlock() self.peersLock.Lock() @@ -110,20 +136,70 @@ func (self *BlockPool) Stop() { return } self.running = false + self.lock.Unlock() poolLogger.Infoln("Stopping") close(self.quit) - self.lock.Lock() + self.wg.Wait() + self.peersLock.Lock() self.peers = nil - self.pool = nil self.peer = nil - self.wg.Wait() - self.lock.Unlock() 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: + case <-time.After(t): + poolLogger.Debugf("completion timeout") + } + + self.flushC = make(chan bool) + + poolLogger.Infoln("processes complete") } @@ -131,29 +207,48 @@ func (self *BlockPool) Stop() { // 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, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) bool { + self.peersLock.Lock() defer self.peersLock.Unlock() - if self.peers[peerId] != nil { - panic("peer already added") + peer, ok := self.peers[peerId] + if ok { + poolLogger.Debugf("update peer %v with td %v and current block %x", peerId, td, currentBlock[:4]) + peer.td = td + peer.currentBlock = currentBlock + } else { + peer = &peerInfo{ + td: td, + currentBlock: currentBlock, + id: peerId, //peer.Identity().Pubkey() + requestBlockHashes: requestBlockHashes, + requestBlocks: requestBlocks, + peerError: peerError, + sections: make(map[string]*section), + } + self.peers[peerId] = peer + poolLogger.Debugf("add new peer %v with td %v and current block %x", peerId, td, currentBlock[:4]) } - peer := &peerInfo{ - td: td, - currentBlock: currentBlock, - id: peerId, //peer.Identity().Pubkey() - requestBlockHashes: requestBlockHashes, - requestBlocks: requestBlocks, - peerError: peerError, + // check peer current head + if self.hasBlock(currentBlock) { + // peer not ahead + return false } - self.peers[peerId] = peer - poolLogger.Debugf("add new peer %v with td %v", peerId, td) + + if self.peer == peer { + // new block update + // peer is already active best peer, request hashes + poolLogger.Debugf("[%s] already the best peer. request hashes from %s", peerId, name(currentBlock)) + peer.requestBlockHashes(currentBlock) + return true + } + currentTD := ethutil.Big0 if self.peer != nil { currentTD = self.peer.td } if td.Cmp(currentTD) > 0 { - self.peer.stop(peer) - peer.start(self.peer) - poolLogger.Debugf("peer %v promoted to best peer", peerId) + poolLogger.Debugf("peer %v promoted best peer", peerId) + self.switchPeer(self.peer, peer) self.peer = peer return true } @@ -164,15 +259,15 @@ func (self *BlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string, func (self *BlockPool) RemovePeer(peerId string) { self.peersLock.Lock() defer self.peersLock.Unlock() - peer := self.peers[peerId] - if peer == nil { + peer, ok := self.peers[peerId] + if !ok { return } - self.peers[peerId] = nil - poolLogger.Debugf("remove peer %v", peerId[0:4]) + delete(self.peers, peerId) + poolLogger.Debugf("remove peer %v", peerId) // if current best peer is removed, need find a better one - if self.peer != nil && peerId == self.peer.id { + if self.peer == peer { var newPeer *peerInfo max := ethutil.Big0 // peer with the highest self-acclaimed TD is chosen @@ -182,16 +277,35 @@ func (self *BlockPool) RemovePeer(peerId string) { newPeer = info } } - self.peer.stop(peer) - peer.start(self.peer) + self.peer = newPeer + self.switchPeer(peer, newPeer) if newPeer != nil { - poolLogger.Debugf("peer %v with td %v promoted to best peer", newPeer.id[0:4], newPeer.td) + poolLogger.Infof("peer %v with td %v promoted to best peer", newPeer.id, newPeer.td) } else { poolLogger.Warnln("no peers left") } } } +func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) { + if newPeer != nil { + entry := self.get(newPeer.currentBlock) + if entry == nil { + poolLogger.Debugf("[%s] head block [%s] not found, requesting hashes", newPeer.id, name(newPeer.currentBlock)) + newPeer.requestBlockHashes(newPeer.currentBlock) + } else { + poolLogger.Debugf("[%s] head block [%s] found, activate chain at section [%s]", newPeer.id, name(newPeer.currentBlock), sectionName(entry.section)) + self.activateChain(entry.section, newPeer) + } + } + if oldPeer != nil { + oldPeer.stop(newPeer) + } + if newPeer != nil { + newPeer.start(oldPeer) + } +} + // 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 @@ -206,160 +320,259 @@ func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) return } // peer is still the best + poolLogger.Debugf("adding hashes for best peer %s", peerId) - var child *poolNode - var depth int - - // iterate using next (rlp stream lazy decoder) feeding hashesC self.wg.Add(1) + self.procWg.Add(1) + go func() { - for { + var size, n int + var hash []byte + var ok bool = true + var section, child, parent *section + var entry *poolEntry + var nodes []*poolNode + + LOOP: + // iterate using next (rlp stream lazy decoder) feeding hashesC + for hash, ok = next(); ok; hash, ok = next() { + n++ select { case <-self.quit: - return + break LOOP case <-peer.quitC: // if the peer is demoted, no more hashes taken - break + break LOOP default: - hash, ok := next() - if !ok { - // message consumed chain skeleton built - break - } - // check if known block connecting the downloaded chain to our blockchain - if self.hasBlock(hash) { - poolLogger.Infof("known block (%x...)\n", hash[0:4]) - if child != nil { - child.Lock() - // mark child as absolute pool root with parent known to blockchain - child.knownParent = true - child.Unlock() - } - break - } - // - var parent *poolNode - // look up node in pool - parent = self.get(hash) - if parent != nil { - // reached a known chain in the pool - // request blocks on the newly added part of the chain - if child != nil { - self.link(parent, child) - - // activate the current chain - self.activateChain(parent, peer, true) - poolLogger.Debugf("potential chain of %v blocks added, reached blockpool, activate chain", depth) - break - } - // if this is the first hash, we expect to find it - parent.RLock() - grandParent := parent.parent - parent.RUnlock() - if grandParent != nil { - // activate the current chain - self.activateChain(parent, peer, true) - poolLogger.Debugf("block hash found, activate chain") - break - } - // the first node is the root of a chain in the pool, rejoice and continue - } - // if node does not exist, create it and index in the pool - section := §ion{} - if child == nil { - section.top = parent - } - parent = &poolNode{ - hash: hash, - child: child, - section: section, - peer: peerId, - } - self.set(hash, parent) - poolLogger.Debugf("create potential block for %x...", hash[0:4]) - - depth++ - child = parent } + if self.hasBlock(hash) { + // check if known block connecting the downloaded chain to our blockchain + poolLogger.Debugf("[%s] known block", name(hash)) + // mark child as absolute pool root with parent known to blockchain + if section != nil { + self.connectToBlockChain(section) + } else { + if child != nil { + self.connectToBlockChain(child) + } + } + break LOOP + } + // look up node in pool + entry = self.get(hash) + if entry != nil { + poolLogger.Debugf("[%s] found block", name(hash)) + // 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.Debugf("[%s] first hash is orphan block, keep building", name(hash)) + child = entry.section + continue LOOP + } + poolLogger.Debugf("[%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 + poolLogger.Debugf("[%s] create node %v", name(hash), size) + node := &poolNode{ + hash: hash, + peer: peerId, + } + if size == 0 { + section = newSection() + } + nodes = append(nodes, node) + size++ + } //for + + self.chainLock.Lock() + poolLogger.Debugf("lock chain lock") + + poolLogger.Debugf("read %v hashes added by %s", n, peerId) + + if parent != nil && entry != nil && entry.node != parent.top { + poolLogger.Debugf("[%s] fork section", sectionName(parent)) + parent.controlC <- false + 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 <- false + close(waiter) } - if child != nil { - poolLogger.Debugf("chain of %v hashes added", depth) - // start a processSection on the last node, but switch off asking - // hashes and blocks until next peer confirms this chain - section := self.processSection(child) - peer.addSection(child.hash, section) - section.start() + + if size > 0 { + self.processSection(section, nodes) + poolLogger.Debugf("[%s]->[%s](%v)->[%s] new chain section", sectionName(parent), sectionName(section), size, sectionName(child)) + self.link(parent, section) + self.link(section, child) + } else { + poolLogger.Debugf("[%s]->[%s] connecting known sections", sectionName(parent), sectionName(child)) + self.link(parent, child) } + + self.chainLock.Unlock() + poolLogger.Debugf("[%s] unlock chain lock", sectionName(section)) + + if parent != nil { + poolLogger.Debugf("[%s] activating parent chain [%s]...", name(parent.top.hash), sectionName(parent)) + self.activateChain(parent, peer) + poolLogger.Debugf("[%s] activated parent chain [%s]. done", name(parent.top.hash), sectionName(parent)) + } + + if section != nil { + poolLogger.Debugf("[%s] activate new section process", sectionName(section)) + peer.addSection(section.top.hash, section) + section.controlC <- true + } + self.procWg.Done() + self.wg.Done() + }() } +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() - node := self.get(hash) - node.RLock() - b := node.block - node.RUnlock() - if b != nil { + poolLogger.Debugf("adding block [%s] by peer %s", name(hash), peerId) + if self.hasBlock(hash) { + poolLogger.Debugf("block [%s] already known", name(hash)) return } - if node == nil && !self.hasBlock(hash) { + entry := self.get(hash) + if entry == nil { + poolLogger.Debugf("unrequested block [%x] by peer %s", hash, peerId) self.peerError(peerId, ErrUnrequestedBlock, "%x", hash) return } + + node := entry.node + node.lock.Lock() + defer node.lock.Unlock() + poolLogger.Debugf("adding block [%s] by peer %s", name(hash), peerId) + + // check if block already present + if node.block != nil { + poolLogger.Debugf("block [%x] already sent by %s", hash, node.blockBy) + return + } + // validate block for PoW if !self.verifyPoW(block) { + poolLogger.Debugf("invalid pow on block [%x] by peer %s", hash, peerId) self.peerError(peerId, ErrInvalidPoW, "%x", hash) + return } - node.Lock() + + poolLogger.Debugf("added block [%s] by peer %s", name(hash), peerId) node.block = block - node.source = peerId - node.Unlock() + node.blockBy = peerId + } -// iterates down a known poolchain and activates fetching processes -// on each chain section for the peer -// stops if the peer is demoted -// registers last section root as root for the peer (in case peer is promoted a second time, to remember) -func (self *BlockPool) activateChain(node *poolNode, peer *peerInfo, on bool) { - self.wg.Add(1) - go func() { - for { - node.sectionRLock() - bottom := node.section.bottom - if bottom == nil { // the chain section is being created or killed - break - } - // register this section with the peer - if peer != nil { - peer.addSection(bottom.hash, bottom.section) - if on { - bottom.section.start() - } else { - bottom.section.start() - } - } - if bottom.parent == nil { - node = bottom - break - } - // if peer demoted stop activation - select { - case <-peer.quitC: - break - default: - } +func (self *BlockPool) connectToBlockChain(section *section) { + section.lock.RLock() + poolLogger.Debugf("connect to blockchain...") + defer section.lock.RUnlock() + if section.off { + self.addSectionToBlockChain(section) + } else { + close(section.blockChainC) + } + poolLogger.Debugf("connect to blockchain done") +} - node = bottom.parent - bottom.sectionRUnlock() +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 } - // remember root for this peer - peer.addRoot(node) - self.wg.Done() - }() + 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.Debugf("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.Debugf("invalid block %x", node.hash) + poolLogger.Debugf("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.Debugf("[%s] activate known chain for peer %s", sectionName(section), peer.id) + i := 0 +LOOP: + for section != nil { + // register this section with the peer + poolLogger.Debugf("[%s] register section with peer %s", sectionName(section), peer.id) + peer.addSection(section.top.hash, section) + poolLogger.Debugf("[%s] activate section process", sectionName(section)) + section.controlC <- true + i++ + // section.lock.RLock() + // parent := section.parent + // section.lock.RUnlock() + // section = parent + poolLogger.Debugf(" before") + section = self.getParent(section) + poolLogger.Debugf(" after") + select { + case <-peer.quitC: + break LOOP + case <-self.quit: + break LOOP + default: + } + } } // main worker thread on each section in the poolchain @@ -370,261 +583,325 @@ func (self *BlockPool) activateChain(node *poolNode, peer *peerInfo, on bool) { // - 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(node *poolNode) *section { - // absolute time after which sub-chain is killed if not complete (some blocks are missing) - suicideTimer := time.After(blockTimeout * time.Minute) - var blocksRequestTimer, blockHashesRequestTimer <-chan time.Time - var nodeC, missingC, processC chan *poolNode - controlC := make(chan bool) - resetC := make(chan bool) - var hashes [][]byte - var i, total, missing, lastMissing, depth int - var blockHashesRequests, blocksRequests int - var idle int - var init, alarm, done, same, running, once bool - orignode := node - hash := node.hash +func (self *BlockPool) processSection(section *section, nodes []*poolNode) { - node.sectionLock() - defer node.sectionUnlock() - section := §ion{controlC: controlC, resetC: resetC} - node.section = section + for i, node := range nodes { + entry := &poolEntry{node: node, section: section, index: i} + self.set(node.hash, entry) + } + section.bottom = nodes[len(nodes)-1] + section.top = nodes[0] + section.nodes = nodes + poolLogger.Debugf("[%s] setup section process", sectionName(section)) + + self.wg.Add(1) go func() { - self.wg.Add(1) + + // absolute time after which sub-chain is killed if not complete (some blocks are missing) + suicideTimer := time.After(blockTimeout * time.Minute) + + 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, total, missing, lastMissing, depth int + var idle int + var init, done, same, running, ready bool + var insertChain bool + + var blockChainC = section.blockChainC + + LOOP: for { - node.sectionRLock() - controlC = node.section.controlC - node.sectionRUnlock() - if init { - // missing blocks read from nodeC - // initialized section - if depth == 0 { - break + if insertChain { + insertChain = false + rest, err := self.addSectionToBlockChain(section) + if err != nil { + close(section.suicideC) + continue LOOP } - // enable select case to read missing block when ready - processC = missingC - missingC = make(chan *poolNode, lastMissing) - nodeC = nil - // only do once - init = false - } else { - if !once { - missingC = nil - processC = nil - i = 0 - total = 0 - lastMissing = 0 + if rest == 0 { + blocksRequestsComplete = true + child := self.getChild(section) + if child != nil { + self.connectToBlockChain(child) + } } } - // went through all blocks in section - if i != 0 && i == lastMissing { - if len(hashes) > 0 { - // send block requests to peers - self.requestBlocks(blocksRequests, hashes) - } - blocksRequests++ - poolLogger.Debugf("[%x] block request attempt %v: missing %v/%v/%v", hash[0:4], blocksRequests, missing, total, depth) - if missing == lastMissing { - // idle round - if same { - // more than once - idle++ - // too many idle rounds - if idle > blocksRequestMaxIdleRounds { - poolLogger.Debugf("[%x] block requests had %v idle rounds (%v total attempts): missing %v/%v/%v\ngiving up...", hash[0:4], idle, blocksRequests, missing, total, depth) - self.killChain(node, nil) - break - } - } else { - idle = 0 - } - same = true + 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(section), 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.Debugf("[%s] got all blocks. process complete (%v total blocksRequests): missing %v/%v/%v", sectionName(section), blocksRequests, missing, total, depth) + blocksRequestsComplete = true + blocksRequestTimer = nil + blocksRequestTime = false } else { - if missing == 0 { - // no missing nodes - poolLogger.Debugf("block request process complete on section %x... (%v total blocksRequests): missing %v/%v/%v", hash[0:4], blockHashesRequests, blocksRequests, missing, total, depth) - node.Lock() - orignode.complete = true - node.Unlock() - blocksRequestTimer = nil - if blockHashesRequestTimer == nil { - // not waiting for hashes any more - poolLogger.Debugf("hash request on root %x... successful (%v total attempts)\nquitting...", hash[0:4], blockHashesRequests) - break - } // otherwise suicide if no hashes coming + // some missing blocks + blocksRequests++ + poolLogger.Debugf("[%s] block request attempt %v: missing %v/%v/%v", sectionName(section), blocksRequests, missing, total, depth) + if len(hashes) > 0 { + // send block requests to peers + self.requestBlocks(blocksRequests, hashes) + hashes = nil + } + poolLogger.Debugf("[%s] check if there is missing blocks", sectionName(section)) + if missing == lastMissing { + // idle round + if same { + // more than once + idle++ + // too many idle rounds + if idle >= blocksRequestMaxIdleRounds { + poolLogger.Debugf("[%s] block requests had %v idle rounds (%v total attempts): missing %v/%v/%v\ngiving up...", sectionName(section), idle, blocksRequests, missing, total, depth) + close(section.suicideC) + } + } else { + idle = 0 + } + same = true + } else { + same = false } - same = false } + poolLogger.Debugf("[%s] done checking missing blocks", sectionName(section)) lastMissing = missing - i = 0 - missing = 0 - // ready for next round - done = true - } - if done && alarm { - poolLogger.Debugf("start checking if new blocks arrived (attempt %v): missing %v/%v/%v", blocksRequests, missing, total, depth) - blocksRequestTimer = time.After(blocksRequestInterval * time.Second) - alarm = false + ready = true done = false - // processC supposed to be empty and never closed so just swap, no need to allocate - tempC := processC - processC = missingC - missingC = tempC + // save a new processC (blocks still missing) + offC = missingC + missingC = processC + // put processC offline + processC = nil + // poolLogger.Debugf("[%s] ready for round %v", sectionName(section), blocksRequests) } - select { - case <-self.quit: - break - case <-suicideTimer: - self.killChain(node, nil) - poolLogger.Warnf("[%x] timeout. (%v total attempts): missing %v/%v/%v", hash[0:4], blocksRequests, missing, total, depth) - break - case <-blocksRequestTimer: - alarm = true - case <-blockHashesRequestTimer: - orignode.RLock() - parent := orignode.parent - orignode.RUnlock() - if parent != nil { + // + + if ready && blocksRequestTime && !blocksRequestsComplete { + poolLogger.Debugf("[%s] check if new blocks arrived (attempt %v): missing %v/%v/%v", sectionName(section), blocksRequests, missing, total, depth) + blocksRequestTimer = time.After(blocksRequestInterval * time.Millisecond) + blocksRequestTime = false + processC = offC + } + + if blockHashesRequestTime { + poolLogger.Debugf("[%s] hash request start", sectionName(section)) + if self.getParent(section) != nil { // if not root of chain, switch off - poolLogger.Debugf("[%x] parent found, hash requests deactivated (after %v total attempts)\n", hash[0:4], blockHashesRequests) + poolLogger.Debugf("[%s] parent found, hash requests deactivated (after %v total attempts)\n", sectionName(section), blockHashesRequests) blockHashesRequestTimer = nil + blockHashesRequestsComplete = true } else { blockHashesRequests++ - poolLogger.Debugf("[%x] hash request on root (%v total attempts)\n", hash[0:4], blockHashesRequests) - self.requestBlockHashes(parent.hash) - blockHashesRequestTimer = time.After(blockHashesRequestInterval * time.Second) + poolLogger.Debugf("[%s] hash request on root (%v total attempts)\n", sectionName(section), blockHashesRequests) + self.requestBlockHashes(section.bottom.hash) + blockHashesRequestTimer = time.After(blockHashesRequestInterval * time.Millisecond) } - case r, ok := <-controlC: - if !ok { - break - } - if running && !r { - poolLogger.Debugf("process on section %x... (%v total attempts): missing %v/%v/%v", hash[0:4], blocksRequests, missing, total, depth) + blockHashesRequestTime = false + poolLogger.Debugf("[%s] hash request done", sectionName(section)) - alarm = false + } + + poolLogger.Debugf("[%s] select", sectionName(section)) + select { + + case <-self.quit: + break LOOP + + case <-self.purgeC: + suicideTimer = time.After(0) + + case <-suicideTimer: + close(section.suicideC) + poolLogger.Debugf("[%s] timeout. (%v total attempts): missing %v/%v/%v", sectionName(section), blocksRequests, missing, total, depth) + + case <-section.suicideC: + poolLogger.Debugf("[%s] suicide", sectionName(section)) + + self.chainLock.Lock() + self.link(nil, section) + self.link(section, nil) + self.chainLock.Unlock() + self.lock.Lock() + for _, node := range section.nodes { + delete(self.pool, string(node.hash)) + } + self.lock.Unlock() + break LOOP + + case <-blocksRequestTimer: + poolLogger.Debugf("[%s] block request time again", sectionName(section)) + blocksRequestTime = true + + case <-blockHashesRequestTimer: + poolLogger.Debugf("[%s] hash request time again", sectionName(section)) + blockHashesRequestTime = true + + case r := <-section.controlC: + + if running && !r { + self.procWg.Done() + poolLogger.Debugf("[%s] idle mode", sectionName(section)) + if init { + poolLogger.Debugf("[%s] off (%v total attempts): missing %v/%v/%v", sectionName(section), blocksRequests, missing, total, depth) + } + + running = false + blocksRequestTime = false blocksRequestTimer = nil + blockHashesRequestTime = false blockHashesRequestTimer = nil - processC = nil + if processC != nil { + offC = processC + processC = nil + } } if !running && r { - poolLogger.Debugf("[%x] on", hash[0:4]) + self.procWg.Add(1) + running = true - orignode.RLock() - parent := orignode.parent - complete := orignode.complete - knownParent := orignode.knownParent - orignode.RUnlock() - if !complete { - poolLogger.Debugf("[%x] activate block requests", hash[0:4]) - blocksRequestTimer = time.After(0) + poolLogger.Debugf("[%s] active mode", sectionName(section)) + poolLogger.Debugf("[%s] check if complete", sectionName(section)) + if !blocksRequestsComplete { + poolLogger.Debugf("[%s] activate block requests", sectionName(section)) + blocksRequestTime = true } - if parent == nil && !knownParent { - // if no parent but not connected to blockchain - poolLogger.Debugf("[%x] activate block hashes requests", hash[0:4]) - blockHashesRequestTimer = time.After(0) - } else { - blockHashesRequestTimer = nil + if !blockHashesRequestsComplete { + poolLogger.Debugf("[%s] activate block hashes requests", sectionName(section)) + blockHashesRequestTime = true } - alarm = true - processC = missingC - if !once { + if !init { // if not run at least once fully, launch iterator - processC = make(chan *poolNode) - missingC = make(chan *poolNode) - self.foldUp(orignode, processC) - once = true + processC = make(chan *poolNode, blockHashesBatchSize) + missingC = make(chan *poolNode, blockHashesBatchSize) + poolLogger.Debugf("[%s] initialise section", sectionName(section)) + i = 0 + missing = 0 + total = 0 + lastMissing = 0 + depth = 0 + self.wg.Add(1) + self.procWg.Add(1) + depth = len(section.nodes) + go func() { + var node *poolNode + IT: + for _, node = range section.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(section)) + processC = offC } } - total = lastMissing - case <-resetC: - once = false + + case waiter := <-section.forkC: + poolLogger.Debugf("[%s] locking for fork", sectionName(section)) + <-waiter + poolLogger.Debugf("[%s] unlocking for fork", sectionName(section)) init = false done = false + ready = false + case node, ok := <-processC: - if !ok { + if !ok && !init { // channel closed, first iteration finished init = true - once = true - continue + done = true + processC = make(chan *poolNode, missing) + + total = missing + + poolLogger.Debugf("[%s] section initalised: missing %v/%v/%v", sectionName(section), missing, total, depth) + continue LOOP } + if ready { + i = 0 + missing = 0 + ready = false + } + poolLogger.Debugf("[%s] process node %v [%x]", sectionName(section), i, node.hash[:4]) i++ // if node has no block - node.RLock() + node.lock.RLock() block := node.block - nhash := node.hash - knownParent := node.knownParent - node.RUnlock() - if !init { - depth++ - } + node.lock.RUnlock() if block == nil { + poolLogger.Debugf("[%s] block missing on [%x]", sectionName(section), node.hash[:4]) missing++ - if !init { - total++ - } - hashes = append(hashes, nhash) + hashes = append(hashes, node.hash) if len(hashes) == blockBatchSize { + poolLogger.Debugf("[%s] request %v missing blocks", sectionName(section), len(hashes)) self.requestBlocks(blocksRequests, hashes) hashes = nil } missingC <- node } else { - // block is found - if knownParent { - // connected to the blockchain, insert the longest chain of blocks - var blocks types.Blocks - child := node - parent := node - node.sectionRLock() - for child != nil && child.block != nil { - parent = child - blocks = append(blocks, parent.block) - child = parent.child - } - node.sectionRUnlock() - poolLogger.Debugf("[%x] insert %v blocks into blockchain", hash[0:4], len(blocks)) - if err := self.insertChain(blocks); err != nil { - // TODO: not clear which peer we need to address - // peerError should dispatch to peer if still connected and disconnect - self.peerError(node.source, ErrInvalidBlock, "%v", err) - poolLogger.Debugf("invalid block %v", node.hash) - poolLogger.Debugf("penalise peers %v (hash), %v (block)", node.peer, node.source) - // penalise peer in node.source - self.killChain(node, nil) - // self.disconnect() - break - } - // if suceeded mark the next one (no block yet) as connected to blockchain - if child != nil { - child.Lock() - child.knownParent = true - child.Unlock() - } - // reset starting node to first node with missing block - orignode = child - // pop the inserted ancestors off the channel - for i := 1; i < len(blocks); i++ { - <-processC - } - // delink inserted chain section - self.killChain(node, parent) + if blockChainC == nil && i == lastMissing { + poolLogger.Debugf("[%s] insert blocks starting from [%s]", sectionName(section), name(node.hash)) + insertChain = true } } - } - } - poolLogger.Debugf("[%x] quit after\n%v block hashes requests\n%v block requests: missing %v/%v/%v", hash[0:4], blockHashesRequests, blocksRequests, missing, total, depth) + poolLogger.Debugf("[%s] %v/%v/%v/%v", sectionName(section), i, missing, total, depth) + if i == lastMissing { + poolLogger.Debugf("[%s] done", sectionName(section)) + 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(section)) + blockChainC = nil + // switch off hash requests in case they were on + blockHashesRequestTime = false + blockHashesRequestTimer = nil + blockHashesRequestsComplete = true + // section root has block + if len(section.nodes) > 0 && section.nodes[len(section.nodes)-1].block != nil { + insertChain = true + } + continue LOOP + + } // select + } // for + poolLogger.Debugf("[%s] quit: %v block hashes requests - %v block requests - missing %v/%v/%v", sectionName(section), blockHashesRequests, blocksRequests, missing, total, depth) + + poolLogger.Debugf("[%s] process complete...", sectionName(section)) + section.lock.Lock() + section.off = true + section.lock.Unlock() + poolLogger.Debugf("[%s] process complete done", sectionName(section)) self.wg.Done() - node.sectionLock() - node.section.controlC = nil - node.sectionUnlock() - // this signals that controller not available + if running { + self.procWg.Done() + } }() - return section - + return } func (self *BlockPool) peerError(peerId string, code int, format string, params ...interface{}) { @@ -640,27 +917,31 @@ func (self *BlockPool) requestBlockHashes(hash []byte) { self.peersLock.Lock() defer self.peersLock.Unlock() if self.peer != nil { + poolLogger.Debugf("request hashes starting on %x from best peer %s", hash[:4], self.peer.id) self.peer.requestBlockHashes(hash) } } func (self *BlockPool) requestBlocks(attempts int, hashes [][]byte) { // distribute block request among known peers + poolLogger.Debugf("request blocks") 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))) - poolLogger.Debugf("request %v missing blocks from %v/%v peers", len(hashes), repetitions, peerCount) i := 0 - indexes := rand.Perm(peerCount)[0:(repetitions - 1)] + 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 from %s", len(hashes), peer.id) peer.requestBlocks(hashes) indexes = indexes[1:] if len(indexes) == 0 { @@ -669,6 +950,8 @@ func (self *BlockPool) requestBlocks(attempts int, hashes [][]byte) { } i++ } + poolLogger.Debugf("done requesting blocks") + } func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) { @@ -679,7 +962,7 @@ func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) { } info, ok := self.peers[peerId] if !ok { - panic("unknown peer") + return nil, false } return info, false } @@ -687,30 +970,16 @@ func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) { func (self *peerInfo) addSection(hash []byte, section *section) { self.lock.Lock() defer self.lock.Unlock() + poolLogger.Debugf("section process %s added to %s", sectionName(section), self.id) self.sections[string(hash)] = section } -func (self *peerInfo) addRoot(node *poolNode) { - self.lock.Lock() - defer self.lock.Unlock() - self.roots = append(self.roots, node) -} - // (re)starts processes registered for this peer (self) func (self *peerInfo) start(peer *peerInfo) { self.lock.Lock() defer self.lock.Unlock() self.quitC = make(chan bool) - for _, root := range self.roots { - root.sectionRLock() - if root.section.bottom != nil { - if root.parent == nil { - self.requestBlockHashes(root.hash) - } - } - root.sectionRUnlock() - } - self.roots = nil + poolLogger.Debugf("[%s] activate section processes", self.id) self.controlSections(peer, true) } @@ -719,6 +988,7 @@ func (self *peerInfo) stop(peer *peerInfo) { self.lock.RLock() defer self.lock.RUnlock() close(self.quitC) + poolLogger.Debugf("[%s] inactivate section processes", self.id) self.controlSections(peer, false) } @@ -727,289 +997,85 @@ func (self *peerInfo) controlSections(peer *peerInfo, on bool) { peer.lock.RLock() defer peer.lock.RUnlock() } - for hash, section := range peer.sections { - if section.done() { + + for hash, section := range self.sections { + + if section.off { + poolLogger.Debugf("[%s][%x] section process complete - remove", self.id, hash[:4]) delete(self.sections, hash) + continue } - _, exists := peer.sections[hash] - if on || peer == nil || exists { + var found bool + if peer != nil { + _, found = peer.sections[hash] + } + + // switch on processes not found in old peer + // and switch off processes not found in new peer + if !found { if on { // self is best peer - section.start() + poolLogger.Debugf("[%s][%s] section process -> active", self.id, sectionName(section)) } else { // (re)starts process without requests, only suicide timer - section.stop() + poolLogger.Debugf("[%s][%s] section process -> inactive", self.id, sectionName(section)) } + section.controlC <- on } } } -// called when parent is found in pool -// parent and child are guaranteed to be on different sections -func (self *BlockPool) link(parent, child *poolNode) { - var top bool - parent.sectionLock() - if child != nil { - child.sectionLock() - } - if parent == parent.section.top && parent.section.top != nil { - top = true - } - var bottom bool +func (self *BlockPool) getParent(sec *section) *section { + poolLogger.Debugf("[") + self.chainLock.RLock() + defer self.chainLock.RUnlock() + poolLogger.Debugf("]") + return sec.parent +} - if child == child.section.bottom { - bottom = true +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 bool, 1), + suicideC: make(chan bool, 1), + blockChainC: make(chan bool, 1), + forkC: make(chan chan bool), } - if parent.child != child { - orphan := parent.child - if orphan != nil { - // got a fork in the chain - if top { - orphan.lock.Lock() - // make old child orphan - orphan.parent = nil - orphan.lock.Unlock() - } else { // we are under section lock - // make old child orphan - orphan.parent = nil - // reset section objects above the fork - nchild := orphan.child - node := orphan - section := §ion{bottom: orphan} - for node.section == nchild.section { - node = nchild - node.section = section - nchild = node.child - } - section.top = node - // set up a suicide - self.processSection(orphan).stop() - } - } else { - // child is on top of a chain need to close section - child.section.bottom = child - } - // adopt new child + return +} + +func (self *BlockPool) link(parent *section, child *section) { + if parent != nil { + exChild := parent.child parent.child = child - if !top { - parent.section.top = parent - // restart section process so that shorter section is scanned for blocks - parent.section.reset() + if exChild != nil && exChild != child { + poolLogger.Debugf("[%s] FORK [%s] -> [%s]", sectionName(parent), sectionName(exChild), sectionName(child)) + exChild.parent = nil } } - if child != nil { - if child.parent != parent { - stepParent := child.parent - if stepParent != nil { - if bottom { - stepParent.Lock() - stepParent.child = nil - stepParent.Unlock() - } else { - // we are on the same section - // if it is a aberrant reverse fork, - stepParent.child = nil - node := stepParent - nparent := stepParent.child - section := §ion{top: stepParent} - for node.section == nparent.section { - node = nparent - node.section = section - node = node.parent - } - } - } else { - // linking to a root node, ie. parent is under the root of a chain - parent.section.top = parent - } + exParent := child.parent + if exParent != nil && exParent != parent { + poolLogger.Debugf("[%s] REV FORK [%s] -> [%s]", sectionName(child), sectionName(exParent), sectionName(parent)) + exParent.child = nil } child.parent = parent - child.section.bottom = child - } - // this needed if someone lied about the parent before - child.knownParent = false - - parent.sectionUnlock() - if child != nil { - child.sectionUnlock() } } -// this immediately kills the chain from node to end (inclusive) section by section -func (self *BlockPool) killChain(node *poolNode, end *poolNode) { - poolLogger.Debugf("kill chain section with root node %v", node) - - node.sectionLock() - node.section.abort() - self.set(node.hash, nil) - child := node.child - top := node.section.top - i := 1 - self.wg.Add(1) - go func() { - var quit bool - for node != top && node != end && child != nil { - node = child - select { - case <-self.quit: - quit = true - break - default: - } - self.set(node.hash, nil) - child = node.child - } - poolLogger.Debugf("killed chain section of %v blocks with root node %v", i, node) - if !quit { - if node == top { - if node != end && child != nil && end != nil { - // - self.killChain(child, end) - } - } else { - if child != nil { - // delink rest of this section if ended midsection - child.section.bottom = child - child.parent = nil - } - } - } - node.section.bottom = nil - node.sectionUnlock() - self.wg.Done() - }() -} - -// structure to store long range links on chain to skip along -type section struct { - lock sync.RWMutex - bottom *poolNode - top *poolNode - controlC chan bool - resetC chan bool -} - -func (self *section) start() { +func (self *BlockPool) get(hash []byte) (node *poolEntry) { self.lock.RLock() defer self.lock.RUnlock() - if self.controlC != nil { - self.controlC <- true - } -} - -func (self *section) stop() { - self.lock.RLock() - defer self.lock.RUnlock() - if self.controlC != nil { - self.controlC <- false - } -} - -func (self *section) reset() { - self.lock.RLock() - defer self.lock.RUnlock() - if self.controlC != nil { - self.resetC <- true - self.controlC <- false - } -} - -func (self *section) abort() { - self.lock.Lock() - defer self.lock.Unlock() - if self.controlC != nil { - close(self.controlC) - self.controlC = nil - } -} - -func (self *section) done() bool { - self.lock.Lock() - defer self.lock.Unlock() - if self.controlC != nil { - return true - } - return false -} - -func (self *BlockPool) get(hash []byte) (node *poolNode) { - self.lock.Lock() - defer self.lock.Unlock() return self.pool[string(hash)] } -func (self *BlockPool) set(hash []byte, node *poolNode) { +func (self *BlockPool) set(hash []byte, node *poolEntry) { self.lock.Lock() defer self.lock.Unlock() self.pool[string(hash)] = node } - -// first time for block request, this iteration retrieves nodes of the chain -// from node up to top (all the way if nil) via child links -// copies the controller -// and feeds nodeC channel -// this is performed under section readlock to prevent top from going away -// when -func (self *BlockPool) foldUp(node *poolNode, nodeC chan *poolNode) { - self.wg.Add(1) - go func() { - node.sectionRLock() - defer node.sectionRUnlock() - for node != nil { - select { - case <-self.quit: - break - case nodeC <- node: - if node == node.section.top { - break - } - node = node.child - } - } - close(nodeC) - self.wg.Done() - }() -} - -func (self *poolNode) Lock() { - self.sectionLock() - self.lock.Lock() -} - -func (self *poolNode) Unlock() { - self.lock.Unlock() - self.sectionUnlock() -} - -func (self *poolNode) RLock() { - self.lock.RLock() -} - -func (self *poolNode) RUnlock() { - self.lock.RUnlock() -} - -func (self *poolNode) sectionLock() { - self.lock.RLock() - defer self.lock.RUnlock() - self.section.lock.Lock() -} - -func (self *poolNode) sectionUnlock() { - self.lock.RLock() - defer self.lock.RUnlock() - self.section.lock.Unlock() -} - -func (self *poolNode) sectionRLock() { - self.lock.RLock() - defer self.lock.RUnlock() - self.section.lock.RLock() -} - -func (self *poolNode) sectionRUnlock() { - self.lock.RLock() - defer self.lock.RUnlock() - self.section.lock.RUnlock() -} diff --git a/eth/block_pool_test.go b/eth/block_pool_test.go index 315cc748db..09392a82be 100644 --- a/eth/block_pool_test.go +++ b/eth/block_pool_test.go @@ -1,115 +1,65 @@ package eth import ( - "bytes" "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/logger" + "github.com/ethereum/go-ethereum/pow" ) -var sys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel)) +const waitTimeout = 60 // seconds -type testChainManager struct { - knownBlock func(hash []byte) bool - addBlock func(*types.Block) error - checkPoW func(*types.Block) bool -} +var logsys = logger.NewStdLogSystem(os.Stdout, log.LstdFlags, logger.LogLevel(logger.DebugLevel)) -func (self *testChainManager) KnownBlock(hash []byte) bool { - if self.knownBlock != nil { - return self.knownBlock(hash) +var ini = false + +func logInit() { + if !ini { + logger.AddLogSystem(logsys) + ini = true } - return false } -func (self *testChainManager) AddBlock(block *types.Block) error { - if self.addBlock != nil { - return self.addBlock(block) - } - return nil -} - -func (self *testChainManager) CheckPoW(block *types.Block) bool { - if self.checkPoW != nil { - return self.checkPoW(block) - } - return false -} - -func knownBlock(hashes ...[]byte) (f func([]byte) bool) { - f = func(block []byte) bool { - for _, hash := range hashes { - if bytes.Compare(block, hash) == 0 { - return true - } - } +// test helpers +func arrayEq(a, b []int) bool { + if len(a) != len(b) { return false } - return -} - -func addBlock(hashes ...[]byte) (f func(*types.Block) error) { - f = func(block *types.Block) error { - for _, hash := range hashes { - if bytes.Compare(block.Hash(), hash) == 0 { - return fmt.Errorf("invalid by test") - } + for i := range a { + if a[i] != b[i] { + return false } - return nil - } - return -} - -func checkPoW(hashes ...[]byte) (f func(*types.Block) bool) { - f = func(block *types.Block) bool { - for _, hash := range hashes { - if bytes.Compare(block.Hash(), hash) == 0 { - return false - } - } - return true - } - return -} - -func newTestChainManager(knownBlocks [][]byte, invalidBlocks [][]byte, invalidPoW [][]byte) *testChainManager { - return &testChainManager{ - knownBlock: knownBlock(knownBlocks...), - addBlock: addBlock(invalidBlocks...), - checkPoW: checkPoW(invalidPoW...), } + 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 newTestBlockPool(knownBlockIndexes []int, invalidBlockIndexes []int, invalidPoWIndexes []int) (hashPool *testHashPool, blockPool *BlockPool) { - hashPool = &testHashPool{make(intToHash), make(hashToInt)} - knownBlocks := hashPool.indexesToHashes(knownBlockIndexes) - invalidBlocks := hashPool.indexesToHashes(invalidBlockIndexes) - invalidPoW := hashPool.indexesToHashes(invalidPoWIndexes) - blockPool = NewBlockPool(newTestChainManager(knownBlocks, invalidBlocks, invalidPoW)) - return -} - 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 { @@ -123,6 +73,8 @@ func (self *testHashPool) indexesToHashes(indexes []int) (hashes [][]byte) { } 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 { @@ -133,66 +85,812 @@ func (self *testHashPool) hashesToIndexes(hashes [][]byte) (indexes []int) { return } -type protocolChecker struct { +// 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 - invalidBlocks []error + blocksRequestsMap map[int]bool + peerErrors []int + blockPool *BlockPool hashPool *testHashPool - lock sync.Mutex + 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) { + i := 0 + 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) + 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) -func (self *protocolChecker) requestBlockHashesCallBack() (requestBlockHashesCallBack func([]byte) error) { - requestBlockHashesCallBack = func(hash []byte) error { - indexes := self.hashPool.hashesToIndexes([][]byte{hash}) - self.lock.Lock() - defer self.lock.Unlock() - self.blockHashesRequests = append(self.blockHashesRequests, indexes[0]) - return nil - } - return +// 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 } -func (self *protocolChecker) requestBlocksCallBack() (requestBlocksCallBack func([][]byte) error) { - requestBlocksCallBack = func(hashes [][]byte) error { - indexes := self.hashPool.hashesToIndexes(hashes) - self.lock.Lock() - defer self.lock.Unlock() - self.blocksRequests = append(self.blocksRequests, indexes) - 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 + return nil } -func (self *protocolChecker) invalidBlockCallBack() (invalidBlockCallBack func(error)) { - invalidBlockCallBack = func(err error) { - self.invalidBlocks = append(self.invalidBlocks, err) - } - return +// 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) { - ethlogger.AddLogSystem(sys) - knownBlockIndexes := []int{0, 1} - invalidBlockIndexes := []int{2, 3} - invalidPoWIndexes := []int{4, 5} - hashPool, blockPool := newTestBlockPool(knownBlockIndexes, invalidBlockIndexes, invalidPoWIndexes) - // TODO: - // hashPool, blockPool, blockChainChecker = newTestBlockPool(knownBlockIndexes, invalidBlockIndexes, invalidPoWIndexes) - peer0 := &protocolChecker{ - // blockHashesRequests: make([]int), - // blocksRequests: make([][]int), - // invalidBlocks: make([]error), - hashPool: hashPool, - } - best := blockPool.AddPeer(ethutil.Big1, newHash(100), "0", - peer0.requestBlockHashesCallBack(), - peer0.requestBlocksCallBack(), - peer0.invalidBlockCallBack(), - ) + 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("peer not accepted as 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.checkBlockHashesRequests(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.checkBlockHashesRequests(2, 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.checkBlockHashesRequests(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") + } + + 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.checkBlockHashesRequests(0, 0, 3) + blockPool.Stop() } + +func TestPeerWithKnownBlock(t *testing.T) { + logInit() + _, blockPool, blockPoolTester := newTestBlockPool(t) + blockPoolTester.refBlockChain[0] = nil + blockPoolTester.blockChain[0] = nil + // hashPool, blockPool, blockPoolTester := newTestBlockPool() + blockPool.Start() + + peer0 := blockPoolTester.newPeer("0", 1, 0) + peer0.AddPeer() + + blockPool.Stop() + // no request on known block + peer0.checkBlockHashesRequests() +} + +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() + go peer1.AddBlockHashes(2, 1, 0) + peer1.AddBlocks(0, 1, 2) + + blockPool.Wait(waitTimeout * time.Second) + blockPool.Stop() + blockPoolTester.refBlockChain[2] = []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.AddBlockHashes(3, 2, 1, 0) + peer1.AddBlocks(0, 1, 2, 3) + + 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") + } +} + +func TestVerifyPoW(t *testing.T) { + 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] == 1 && !first { + first = true + return false + } else { + return true + } + + } + + blockPool.Start() + + peer1 := blockPoolTester.newPeer("peer1", 1, 2) + peer1.AddPeer() + go peer1.AddBlockHashes(2, 1, 0) + peer1.AddBlocks(0, 1, 2) + peer1.AddBlocks(0, 1) + + blockPool.Wait(waitTimeout * time.Second) + blockPool.Stop() + blockPoolTester.refBlockChain[2] = []int{} + 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.AddBlockHashes(5, 4, 3) + go peer1.AddBlocks(2, 3, 4, 5) + 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.AddBlockHashes(5, 4, 3) + peer1.AddBlocks(2, 3) // partially complete section + // peer1 found new blocks + peer1.td = 2 + peer1.currentBlock = 7 + peer1.AddPeer() + go peer1.AddBlockHashes(7, 6, 5) + go peer1.AddBlocks(3, 4, 5, 6, 7) + 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 TestPeerSwitch(t *testing.T) { + logInit() + _, blockPool, blockPoolTester := newTestBlockPool(t) + blockPoolTester.blockChain[0] = nil + blockPoolTester.initRefBlockChain(6) + + blockPool.Start() + + peer1 := blockPoolTester.newPeer("peer1", 1, 5) + peer2 := blockPoolTester.newPeer("peer2", 2, 6) + peer2.blocksRequestsMap = peer1.blocksRequestsMap + + peer1.AddPeer() + go peer1.AddBlockHashes(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.AddBlockHashes(6, 5) // + go peer2.AddBlocks(4, 5, 6) // 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[6] = []int{} + blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain) +} + +func TestPeerDownSwitch(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() + go peer2.AddBlockHashes(6, 5, 4) + peer2.AddBlocks(5, 6) // partially complete, section will be preserved + 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, 5) // tests that section set by demoted peer is remembered and blocks are accepted + 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.AddBlockHashes(8, 7, 6) + go peer2.AddBlockHashes(6, 5, 4) + peer2.AddBlocks(5, 6) // section partially complete + peer1.AddPeer() // peer1 is promoted as best peer + go 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.AddBlockHashes(9, 8, 7, 3, 2) + peer1.AddBlocks(1, 2, 3, 7, 8, 9) + peer2.AddPeer() // peer2 is promoted as best peer + go peer2.AddBlockHashes(6, 5, 4, 3, 2) // fork on 3 -> 4 (earlier child: 7) + go peer2.AddBlocks(1, 2, 3, 4, 5, 6) + go peer2.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() + go peer1.AddBlockHashes(9, 8, 7, 3, 2) + peer1.AddBlocks(8, 9) // partial section + peer2.AddPeer() // + go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3 + peer2.AddBlocks(1, 2, 3, 4, 5, 6) // + + // peer1 finds new blocks + peer1.td = 3 + peer1.currentBlock = 11 + peer1.AddPeer() + go peer1.AddBlockHashes(11, 10, 9) + peer1.AddBlocks(7, 8, 9, 10, 11) + go peer1.AddBlockHashes(7, 3) // tests that hash request from fork root is remembered + go peer1.AddBlocks(3, 7) // tests that block requests on earlier fork are remembered + // go peer1.AddBlockHashes(1, 0) // tests that hash request from root of connecting chain section (added by demoted peer) is remembered + go peer1.AddBlockHashes(2, 1, 0) // tests that hash request from root of connecting chain section (added by demoted peer) is remembered + peer1.AddBlocks(0, 1, 2, 3) + + 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.AddBlockHashes(9, 8, 7, 3, 2) + peer1.AddBlocks(8, 9) + peer2.AddPeer() // + go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3 + peer2.AddBlocks(2, 3, 4, 5, 6) // + blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer + peer1.AddBlockHashes(7, 3) // tests that hash request from fork root is remembered + go peer1.AddBlocks(3, 7, 8) // tests that block requests on earlier fork are remembered + go peer1.AddBlockHashes(2, 1, 0) // + peer1.AddBlocks(0, 1, 2, 3) + + 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.AddBlockHashes(9, 8, 7) + peer1.AddBlocks(3, 7, 8, 9) // 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 + peer2.AddPeer() // + go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3 + peer2.AddBlocks(2, 3, 4, 5, 6) // 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) + +}