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] by peer %s", name(hash), peerId) self.peerError(peerId, ErrInvalidPoW, "%x", hash) return } } poolLogger.Debugf("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 }