aboutsummaryrefslogtreecommitdiffstats
path: root/eth/block_pool.go
diff options
context:
space:
mode:
Diffstat (limited to 'eth/block_pool.go')
-rw-r--r--eth/block_pool.go1239
1 files changed, 0 insertions, 1239 deletions
diff --git a/eth/block_pool.go b/eth/block_pool.go
deleted file mode 100644
index 13016c694..000000000
--- a/eth/block_pool.go
+++ /dev/null
@@ -1,1239 +0,0 @@
-package eth
-
-import (
- "bytes"
- "fmt"
- "math"
- "math/big"
- "math/rand"
- "sort"
- "sync"
- "time"
-
- "github.com/ethereum/go-ethereum/core/types"
- "github.com/ethereum/go-ethereum/ethutil"
- ethlogger "github.com/ethereum/go-ethereum/logger"
- "github.com/ethereum/go-ethereum/pow"
-)
-
-var poolLogger = ethlogger.NewLogger("Blockpool")
-
-const (
- blockHashesBatchSize = 256
- blockBatchSize = 64
- blocksRequestInterval = 500 // ms
- blocksRequestRepetition = 1
- blockHashesRequestInterval = 500 // ms
- blocksRequestMaxIdleRounds = 100
- blockHashesTimeout = 60 // seconds
- blocksTimeout = 120 // seconds
-)
-
-type poolNode struct {
- lock sync.RWMutex
- hash []byte
- td *big.Int
- block *types.Block
- parent *poolNode
- peer string
- blockBy string
-}
-
-type poolEntry struct {
- node *poolNode
- section *section
- index int
-}
-
-type BlockPool struct {
- lock sync.RWMutex
- chainLock sync.RWMutex
-
- pool map[string]*poolEntry
-
- peersLock sync.RWMutex
- peers map[string]*peerInfo
- peer *peerInfo
-
- quit chan bool
- purgeC chan bool
- flushC chan bool
- wg sync.WaitGroup
- procWg sync.WaitGroup
- running bool
-
- // the minimal interface with blockchain
- hasBlock func(hash []byte) bool
- insertChain func(types.Blocks) error
- verifyPoW func(pow.Block) bool
-}
-
-type peerInfo struct {
- lock sync.RWMutex
-
- td *big.Int
- currentBlockHash []byte
- currentBlock *types.Block
- currentBlockC chan *types.Block
- parentHash []byte
- headSection *section
- headSectionC chan *section
- id string
-
- requestBlockHashes func([]byte) error
- requestBlocks func([][]byte) error
- peerError func(int, string, ...interface{})
-
- sections map[string]*section
-
- quitC chan bool
-}
-
-// structure to store long range links on chain to skip along
-type section struct {
- lock sync.RWMutex
- parent *section
- child *section
- top *poolNode
- bottom *poolNode
- nodes []*poolNode
- controlC chan *peerInfo
- suicideC chan bool
- blockChainC chan bool
- forkC chan chan bool
- offC chan bool
-}
-
-func NewBlockPool(hasBlock func(hash []byte) bool, insertChain func(types.Blocks) error, verifyPoW func(pow.Block) bool,
-) *BlockPool {
- return &BlockPool{
- hasBlock: hasBlock,
- insertChain: insertChain,
- verifyPoW: verifyPoW,
- }
-}
-
-// allows restart
-func (self *BlockPool) Start() {
- self.lock.Lock()
- if self.running {
- self.lock.Unlock()
- return
- }
- self.running = true
- self.quit = make(chan bool)
- self.flushC = make(chan bool)
- self.pool = make(map[string]*poolEntry)
-
- self.lock.Unlock()
-
- self.peersLock.Lock()
- self.peers = make(map[string]*peerInfo)
- self.peersLock.Unlock()
-
- poolLogger.Infoln("Started")
-
-}
-
-func (self *BlockPool) Stop() {
- self.lock.Lock()
- if !self.running {
- self.lock.Unlock()
- return
- }
- self.running = false
-
- self.lock.Unlock()
-
- poolLogger.Infoln("Stopping...")
-
- close(self.quit)
- //self.wg.Wait()
-
- self.peersLock.Lock()
- self.peers = nil
- self.peer = nil
- self.peersLock.Unlock()
-
- self.lock.Lock()
- self.pool = nil
- self.lock.Unlock()
-
- poolLogger.Infoln("Stopped")
-}
-
-func (self *BlockPool) Purge() {
- self.lock.Lock()
- if !self.running {
- self.lock.Unlock()
- return
- }
- self.lock.Unlock()
-
- poolLogger.Infoln("Purging...")
-
- close(self.purgeC)
- self.wg.Wait()
-
- self.purgeC = make(chan bool)
-
- poolLogger.Infoln("Stopped")
-
-}
-
-func (self *BlockPool) Wait(t time.Duration) {
- self.lock.Lock()
- if !self.running {
- self.lock.Unlock()
- return
- }
- self.lock.Unlock()
-
- poolLogger.Infoln("Waiting for processes to complete...")
- close(self.flushC)
- w := make(chan bool)
- go func() {
- self.procWg.Wait()
- close(w)
- }()
-
- select {
- case <-w:
- poolLogger.Infoln("Processes complete")
- case <-time.After(t):
- poolLogger.Warnf("Timeout")
- }
- self.flushC = make(chan bool)
-}
-
-// AddPeer is called by the eth protocol instance running on the peer after
-// the status message has been received with total difficulty and current block hash
-// AddPeer can only be used once, RemovePeer needs to be called when the peer disconnects
-func (self *BlockPool) AddPeer(td *big.Int, currentBlockHash []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) (best bool) {
-
- self.peersLock.Lock()
- defer self.peersLock.Unlock()
- peer, ok := self.peers[peerId]
- if ok {
- if bytes.Compare(peer.currentBlockHash, currentBlockHash) != 0 {
- poolLogger.Debugf("Update peer %v with td %v and current block %s", peerId, td, name(currentBlockHash))
- peer.lock.Lock()
- peer.td = td
- peer.currentBlockHash = currentBlockHash
- peer.currentBlock = nil
- peer.parentHash = nil
- peer.headSection = nil
- peer.lock.Unlock()
- }
- } else {
- peer = &peerInfo{
- td: td,
- currentBlockHash: currentBlockHash,
- id: peerId, //peer.Identity().Pubkey()
- requestBlockHashes: requestBlockHashes,
- requestBlocks: requestBlocks,
- peerError: peerError,
- sections: make(map[string]*section),
- currentBlockC: make(chan *types.Block),
- headSectionC: make(chan *section),
- }
- self.peers[peerId] = peer
- poolLogger.Debugf("add new peer %v with td %v and current block %x", peerId, td, currentBlockHash[:4])
- }
- // check peer current head
- if self.hasBlock(currentBlockHash) {
- // peer not ahead
- return false
- }
-
- if self.peer == peer {
- // new block update
- // peer is already active best peer, request hashes
- poolLogger.Debugf("[%s] already the best peer. Request new head section info from %s", peerId, name(currentBlockHash))
- peer.headSectionC <- nil
- best = true
- } else {
- currentTD := ethutil.Big0
- if self.peer != nil {
- currentTD = self.peer.td
- }
- if td.Cmp(currentTD) > 0 {
- poolLogger.Debugf("peer %v promoted best peer", peerId)
- self.switchPeer(self.peer, peer)
- self.peer = peer
- best = true
- }
- }
- return
-}
-
-func (self *BlockPool) requestHeadSection(peer *peerInfo) {
- self.wg.Add(1)
- self.procWg.Add(1)
- poolLogger.Debugf("[%s] head section at [%s] requesting info", peer.id, name(peer.currentBlockHash))
-
- go func() {
- var idle bool
- peer.lock.RLock()
- quitC := peer.quitC
- currentBlockHash := peer.currentBlockHash
- peer.lock.RUnlock()
- blockHashesRequestTimer := time.NewTimer(0)
- blocksRequestTimer := time.NewTimer(0)
- suicide := time.NewTimer(blockHashesTimeout * time.Second)
- blockHashesRequestTimer.Stop()
- defer blockHashesRequestTimer.Stop()
- defer blocksRequestTimer.Stop()
-
- entry := self.get(currentBlockHash)
- if entry != nil {
- entry.node.lock.RLock()
- currentBlock := entry.node.block
- entry.node.lock.RUnlock()
- if currentBlock != nil {
- peer.lock.Lock()
- peer.currentBlock = currentBlock
- peer.parentHash = currentBlock.ParentHash()
- poolLogger.Debugf("[%s] head block [%s] found", peer.id, name(currentBlockHash))
- peer.lock.Unlock()
- blockHashesRequestTimer.Reset(0)
- blocksRequestTimer.Stop()
- }
- }
-
- LOOP:
- for {
-
- select {
- case <-self.quit:
- break LOOP
-
- case <-quitC:
- poolLogger.Debugf("[%s] head section at [%s] incomplete - quit request loop", peer.id, name(currentBlockHash))
- break LOOP
-
- case headSection := <-peer.headSectionC:
- peer.lock.Lock()
- peer.headSection = headSection
- if headSection == nil {
- oldBlockHash := currentBlockHash
- currentBlockHash = peer.currentBlockHash
- poolLogger.Debugf("[%s] head section changed [%s] -> [%s]", peer.id, name(oldBlockHash), name(currentBlockHash))
- if idle {
- idle = false
- suicide.Reset(blockHashesTimeout * time.Second)
- self.procWg.Add(1)
- }
- blocksRequestTimer.Reset(blocksRequestInterval * time.Millisecond)
- } else {
- poolLogger.DebugDetailf("[%s] head section at [%s] created", peer.id, name(currentBlockHash))
- if !idle {
- idle = true
- suicide.Stop()
- self.procWg.Done()
- }
- }
- peer.lock.Unlock()
- blockHashesRequestTimer.Stop()
-
- case <-blockHashesRequestTimer.C:
- poolLogger.DebugDetailf("[%s] head section at [%s] not found, requesting block hashes", peer.id, name(currentBlockHash))
- peer.requestBlockHashes(currentBlockHash)
- blockHashesRequestTimer.Reset(blockHashesRequestInterval * time.Millisecond)
-
- case currentBlock := <-peer.currentBlockC:
- peer.lock.Lock()
- peer.currentBlock = currentBlock
- peer.parentHash = currentBlock.ParentHash()
- poolLogger.DebugDetailf("[%s] head block [%s] found", peer.id, name(currentBlockHash))
- peer.lock.Unlock()
- if self.hasBlock(currentBlock.ParentHash()) {
- if err := self.insertChain(types.Blocks([]*types.Block{currentBlock})); err != nil {
- peer.peerError(ErrInvalidBlock, "%v", err)
- }
- if !idle {
- idle = true
- suicide.Stop()
- self.procWg.Done()
- }
- } else {
- blockHashesRequestTimer.Reset(0)
- }
- blocksRequestTimer.Stop()
-
- case <-blocksRequestTimer.C:
- peer.lock.RLock()
- poolLogger.DebugDetailf("[%s] head block [%s] not found, requesting", peer.id, name(currentBlockHash))
- peer.requestBlocks([][]byte{peer.currentBlockHash})
- peer.lock.RUnlock()
- blocksRequestTimer.Reset(blocksRequestInterval * time.Millisecond)
-
- case <-suicide.C:
- peer.peerError(ErrInsufficientChainInfo, "peer failed to provide block hashes or head block for block hash %x", currentBlockHash)
- break LOOP
- }
- }
- self.wg.Done()
- if !idle {
- self.procWg.Done()
- }
- }()
-}
-
-// RemovePeer is called by the eth protocol when the peer disconnects
-func (self *BlockPool) RemovePeer(peerId string) {
- self.peersLock.Lock()
- defer self.peersLock.Unlock()
- peer, ok := self.peers[peerId]
- if !ok {
- return
- }
- delete(self.peers, peerId)
- poolLogger.Debugf("remove peer %v", peerId)
-
- // if current best peer is removed, need find a better one
- if self.peer == peer {
- var newPeer *peerInfo
- max := ethutil.Big0
- // peer with the highest self-acclaimed TD is chosen
- for _, info := range self.peers {
- if info.td.Cmp(max) > 0 {
- max = info.td
- newPeer = info
- }
- }
- if newPeer != nil {
- poolLogger.Debugf("peer %v with td %v promoted to best peer", newPeer.id, newPeer.td)
- } else {
- poolLogger.Warnln("no peers")
- }
- self.peer = newPeer
- self.switchPeer(peer, newPeer)
- }
-}
-
-// Entry point for eth protocol to add block hashes received via BlockHashesMsg
-// only hashes from the best peer is handled
-// this method is always responsible to initiate further hash requests until
-// a known parent is reached unless cancelled by a peerChange event
-// this process also launches all request processes on each chain section
-// this function needs to run asynchronously for one peer since the message is discarded???
-func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) {
-
- // register with peer manager loop
-
- peer, best := self.getPeer(peerId)
- if !best {
- return
- }
- // peer is still the best
-
- var size, n int
- var hash []byte
- var ok, headSection bool
- var sec, child, parent *section
- var entry *poolEntry
- var nodes []*poolNode
- bestPeer := peer
-
- hash, ok = next()
- peer.lock.Lock()
- if bytes.Compare(peer.parentHash, hash) == 0 {
- if self.hasBlock(peer.currentBlockHash) {
- return
- }
- poolLogger.Debugf("adding hashes at chain head for best peer %s starting from [%s]", peerId, name(peer.currentBlockHash))
- headSection = true
-
- if entry := self.get(peer.currentBlockHash); entry == nil {
- node := &poolNode{
- hash: peer.currentBlockHash,
- block: peer.currentBlock,
- peer: peerId,
- blockBy: peerId,
- }
- if size == 0 {
- sec = newSection()
- }
- nodes = append(nodes, node)
- size++
- n++
- } else {
- child = entry.section
- }
- } else {
- poolLogger.Debugf("adding hashes for best peer %s starting from [%s]", peerId, name(hash))
- }
- quitC := peer.quitC
- peer.lock.Unlock()
-
-LOOP:
- // iterate using next (rlp stream lazy decoder) feeding hashesC
- for ; ok; hash, ok = next() {
- n++
- select {
- case <-self.quit:
- return
- case <-quitC:
- // if the peer is demoted, no more hashes taken
- bestPeer = nil
- break LOOP
- default:
- }
- if self.hasBlock(hash) {
- // check if known block connecting the downloaded chain to our blockchain
- poolLogger.DebugDetailf("[%s] known block", name(hash))
- // mark child as absolute pool root with parent known to blockchain
- if sec != nil {
- self.connectToBlockChain(sec)
- } else {
- if child != nil {
- self.connectToBlockChain(child)
- }
- }
- break LOOP
- }
- // look up node in pool
- entry = self.get(hash)
- if entry != nil {
- // reached a known chain in the pool
- if entry.node == entry.section.bottom && n == 1 {
- // the first block hash received is an orphan in the pool, so rejoice and continue
- poolLogger.DebugDetailf("[%s] connecting child section", sectionName(entry.section))
- child = entry.section
- continue LOOP
- }
- poolLogger.DebugDetailf("[%s] reached blockpool chain", name(hash))
- parent = entry.section
- break LOOP
- }
- // if node for block hash does not exist, create it and index in the pool
- node := &poolNode{
- hash: hash,
- peer: peerId,
- }
- if size == 0 {
- sec = newSection()
- }
- nodes = append(nodes, node)
- size++
- } //for
-
- self.chainLock.Lock()
-
- poolLogger.DebugDetailf("added %v hashes sent by %s", n, peerId)
-
- if parent != nil && entry != nil && entry.node != parent.top {
- poolLogger.DebugDetailf("[%s] split section at fork", sectionName(parent))
- parent.controlC <- nil
- waiter := make(chan bool)
- parent.forkC <- waiter
- chain := parent.nodes
- parent.nodes = chain[entry.index:]
- parent.top = parent.nodes[0]
- orphan := newSection()
- self.link(orphan, parent.child)
- self.processSection(orphan, chain[0:entry.index])
- orphan.controlC <- nil
- close(waiter)
- }
-
- if size > 0 {
- self.processSection(sec, nodes)
- poolLogger.DebugDetailf("[%s]->[%s](%v)->[%s] new chain section", sectionName(parent), sectionName(sec), size, sectionName(child))
- self.link(parent, sec)
- self.link(sec, child)
- } else {
- poolLogger.DebugDetailf("[%s]->[%s] connecting known sections", sectionName(parent), sectionName(child))
- self.link(parent, child)
- }
-
- self.chainLock.Unlock()
-
- if parent != nil && bestPeer != nil {
- self.activateChain(parent, peer)
- poolLogger.Debugf("[%s] activate parent section [%s]", name(parent.top.hash), sectionName(parent))
- }
-
- if sec != nil {
- peer.addSection(sec.top.hash, sec)
- // request next section here once, only repeat if bottom block arrives,
- // otherwise no way to check if it arrived
- peer.requestBlockHashes(sec.bottom.hash)
- sec.controlC <- bestPeer
- poolLogger.Debugf("[%s] activate new section", sectionName(sec))
- }
-
- if headSection {
- var headSec *section
- switch {
- case sec != nil:
- headSec = sec
- case child != nil:
- headSec = child
- default:
- headSec = parent
- }
- peer.headSectionC <- headSec
- }
-}
-
-func name(hash []byte) (name string) {
- if hash == nil {
- name = ""
- } else {
- name = fmt.Sprintf("%x", hash[:4])
- }
- return
-}
-
-func sectionName(section *section) (name string) {
- if section == nil {
- name = ""
- } else {
- name = fmt.Sprintf("%x-%x", section.bottom.hash[:4], section.top.hash[:4])
- }
- return
-}
-
-// AddBlock is the entry point for the eth protocol when blockmsg is received upon requests
-// It has a strict interpretation of the protocol in that if the block received has not been requested, it results in an error (which can be ignored)
-// block is checked for PoW
-// only the first PoW-valid block for a hash is considered legit
-func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
- hash := block.Hash()
- self.peersLock.Lock()
- peer := self.peer
- self.peersLock.Unlock()
-
- entry := self.get(hash)
- if bytes.Compare(hash, peer.currentBlockHash) == 0 {
- poolLogger.Debugf("add head block [%s] for peer %s", name(hash), peerId)
- peer.currentBlockC <- block
- } else {
- if entry == nil {
- poolLogger.Warnf("unrequested block [%s] by peer %s", name(hash), peerId)
- self.peerError(peerId, ErrUnrequestedBlock, "%x", hash)
- }
- }
- if entry == nil {
- return
- }
-
- node := entry.node
- node.lock.Lock()
- defer node.lock.Unlock()
-
- // check if block already present
- if node.block != nil {
- poolLogger.DebugDetailf("block [%s] already sent by %s", name(hash), node.blockBy)
- return
- }
-
- if self.hasBlock(hash) {
- poolLogger.DebugDetailf("block [%s] already known", name(hash))
- } else {
-
- // validate block for PoW
- if !self.verifyPoW(block) {
- poolLogger.Warnf("invalid pow on block [%s %v] by peer %s", name(hash), block.Number(), peerId)
- self.peerError(peerId, ErrInvalidPoW, "%x", hash)
- return
- }
- }
- poolLogger.DebugDetailf("added block [%s] sent by peer %s", name(hash), peerId)
- node.block = block
- node.blockBy = peerId
-
-}
-
-func (self *BlockPool) connectToBlockChain(section *section) {
- select {
- case <-section.offC:
- self.addSectionToBlockChain(section)
- case <-section.blockChainC:
- default:
- close(section.blockChainC)
- }
-}
-
-func (self *BlockPool) addSectionToBlockChain(section *section) (rest int, err error) {
-
- var blocks types.Blocks
- var node *poolNode
- var keys []string
- rest = len(section.nodes)
- for rest > 0 {
- rest--
- node = section.nodes[rest]
- node.lock.RLock()
- block := node.block
- node.lock.RUnlock()
- if block == nil {
- break
- }
- keys = append(keys, string(node.hash))
- blocks = append(blocks, block)
- }
-
- self.lock.Lock()
- for _, key := range keys {
- delete(self.pool, key)
- }
- self.lock.Unlock()
-
- poolLogger.Infof("insert %v blocks into blockchain", len(blocks))
- err = self.insertChain(blocks)
- if err != nil {
- // TODO: not clear which peer we need to address
- // peerError should dispatch to peer if still connected and disconnect
- self.peerError(node.blockBy, ErrInvalidBlock, "%v", err)
- poolLogger.Warnf("invalid block %x", node.hash)
- poolLogger.Warnf("penalise peers %v (hash), %v (block)", node.peer, node.blockBy)
- // penalise peer in node.blockBy
- // self.disconnect()
- }
- return
-}
-
-func (self *BlockPool) activateChain(section *section, peer *peerInfo) {
- poolLogger.DebugDetailf("[%s] activate known chain for peer %s", sectionName(section), peer.id)
- i := 0
-LOOP:
- for section != nil {
- // register this section with the peer and quit if registered
- poolLogger.DebugDetailf("[%s] register section with peer %s", sectionName(section), peer.id)
- if peer.addSection(section.top.hash, section) == section {
- return
- }
- poolLogger.DebugDetailf("[%s] activate section process", sectionName(section))
- select {
- case section.controlC <- peer:
- case <-section.offC:
- }
- i++
- section = self.getParent(section)
- select {
- case <-peer.quitC:
- break LOOP
- case <-self.quit:
- break LOOP
- default:
- }
- }
-}
-
-// main worker thread on each section in the poolchain
-// - kills the section if there are blocks missing after an absolute time
-// - kills the section if there are maxIdleRounds of idle rounds of block requests with no response
-// - periodically polls the chain section for missing blocks which are then requested from peers
-// - registers the process controller on the peer so that if the peer is promoted as best peer the second time (after a disconnect of a better one), all active processes are switched back on unless they expire and killed ()
-// - when turned off (if peer disconnects and new peer connects with alternative chain), no blockrequests are made but absolute expiry timer is ticking
-// - when turned back on it recursively calls itself on the root of the next chain section
-// - when exits, signals to
-func (self *BlockPool) processSection(sec *section, nodes []*poolNode) {
-
- for i, node := range nodes {
- entry := &poolEntry{node: node, section: sec, index: i}
- self.set(node.hash, entry)
- }
-
- sec.bottom = nodes[len(nodes)-1]
- sec.top = nodes[0]
- sec.nodes = nodes
- poolLogger.DebugDetailf("[%s] setup section process", sectionName(sec))
-
- self.wg.Add(1)
- go func() {
-
- // absolute time after which sub-chain is killed if not complete (some blocks are missing)
- suicideTimer := time.After(blocksTimeout * time.Second)
-
- var peer, newPeer *peerInfo
-
- var blocksRequestTimer, blockHashesRequestTimer <-chan time.Time
- var blocksRequestTime, blockHashesRequestTime bool
- var blocksRequests, blockHashesRequests int
- var blocksRequestsComplete, blockHashesRequestsComplete bool
-
- // node channels for the section
- var missingC, processC, offC chan *poolNode
- // container for missing block hashes
- var hashes [][]byte
-
- var i, missing, lastMissing, depth int
- var idle int
- var init, done, same, ready bool
- var insertChain bool
- var quitC chan bool
-
- var blockChainC = sec.blockChainC
-
- var parentHash []byte
-
- LOOP:
- for {
-
- if insertChain {
- insertChain = false
- rest, err := self.addSectionToBlockChain(sec)
- if err != nil {
- close(sec.suicideC)
- continue LOOP
- }
- if rest == 0 {
- blocksRequestsComplete = true
- child := self.getChild(sec)
- if child != nil {
- self.connectToBlockChain(child)
- }
- }
- }
-
- if blockHashesRequestsComplete && blocksRequestsComplete {
- // not waiting for hashes any more
- poolLogger.Debugf("[%s] section complete %v blocks retrieved (%v attempts), hash requests complete on root (%v attempts)", sectionName(sec), depth, blocksRequests, blockHashesRequests)
- break LOOP
- } // otherwise suicide if no hashes coming
-
- if done {
- // went through all blocks in section
- if missing == 0 {
- // no missing blocks
- poolLogger.DebugDetailf("[%s] got all blocks. process complete (%v total blocksRequests): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
- blocksRequestsComplete = true
- blocksRequestTimer = nil
- blocksRequestTime = false
- } else {
- poolLogger.DebugDetailf("[%s] section checked: missing %v/%v/%v", sectionName(sec), missing, lastMissing, depth)
- // some missing blocks
- blocksRequests++
- if len(hashes) > 0 {
- // send block requests to peers
- self.requestBlocks(blocksRequests, hashes)
- hashes = nil
- }
- if missing == lastMissing {
- // idle round
- if same {
- // more than once
- idle++
- // too many idle rounds
- if idle >= blocksRequestMaxIdleRounds {
- poolLogger.DebugDetailf("[%s] block requests had %v idle rounds (%v total attempts): missing %v/%v/%v\ngiving up...", sectionName(sec), idle, blocksRequests, missing, lastMissing, depth)
- close(sec.suicideC)
- }
- } else {
- idle = 0
- }
- same = true
- } else {
- same = false
- }
- }
- lastMissing = missing
- ready = true
- done = false
- // save a new processC (blocks still missing)
- offC = missingC
- missingC = processC
- // put processC offline
- processC = nil
- }
- //
-
- if ready && blocksRequestTime && !blocksRequestsComplete {
- poolLogger.DebugDetailf("[%s] check if new blocks arrived (attempt %v): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
- blocksRequestTimer = time.After(blocksRequestInterval * time.Millisecond)
- blocksRequestTime = false
- processC = offC
- }
-
- if blockHashesRequestTime {
- var parentSection = self.getParent(sec)
- if parentSection == nil {
- if parent := self.get(parentHash); parent != nil {
- parentSection = parent.section
- self.chainLock.Lock()
- self.link(parentSection, sec)
- self.chainLock.Unlock()
- } else {
- if self.hasBlock(parentHash) {
- insertChain = true
- blockHashesRequestTime = false
- blockHashesRequestTimer = nil
- blockHashesRequestsComplete = true
- continue LOOP
- }
- }
- }
- if parentSection != nil {
- // if not root of chain, switch off
- poolLogger.DebugDetailf("[%s] parent found, hash requests deactivated (after %v total attempts)\n", sectionName(sec), blockHashesRequests)
- blockHashesRequestTimer = nil
- blockHashesRequestsComplete = true
- } else {
- blockHashesRequests++
- poolLogger.Debugf("[%s] hash request on root (%v total attempts)\n", sectionName(sec), blockHashesRequests)
- peer.requestBlockHashes(sec.bottom.hash)
- blockHashesRequestTimer = time.After(blockHashesRequestInterval * time.Millisecond)
- }
- blockHashesRequestTime = false
- }
-
- select {
- case <-self.quit:
- break LOOP
-
- case <-quitC:
- // peer quit or demoted, put section in idle mode
- quitC = nil
- go func() {
- sec.controlC <- nil
- }()
-
- case <-self.purgeC:
- suicideTimer = time.After(0)
-
- case <-suicideTimer:
- close(sec.suicideC)
- poolLogger.Debugf("[%s] timeout. (%v total attempts): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
-
- case <-sec.suicideC:
- poolLogger.Debugf("[%s] suicide", sectionName(sec))
-
- // first delink from child and parent under chainlock
- self.chainLock.Lock()
- self.link(nil, sec)
- self.link(sec, nil)
- self.chainLock.Unlock()
- // delete node entries from pool index under pool lock
- self.lock.Lock()
- for _, node := range sec.nodes {
- delete(self.pool, string(node.hash))
- }
- self.lock.Unlock()
-
- break LOOP
-
- case <-blocksRequestTimer:
- poolLogger.DebugDetailf("[%s] block request time", sectionName(sec))
- blocksRequestTime = true
-
- case <-blockHashesRequestTimer:
- poolLogger.DebugDetailf("[%s] hash request time", sectionName(sec))
- blockHashesRequestTime = true
-
- case newPeer = <-sec.controlC:
-
- // active -> idle
- if peer != nil && newPeer == nil {
- self.procWg.Done()
- if init {
- poolLogger.Debugf("[%s] idle mode (%v total attempts): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth)
- }
- blocksRequestTime = false
- blocksRequestTimer = nil
- blockHashesRequestTime = false
- blockHashesRequestTimer = nil
- if processC != nil {
- offC = processC
- processC = nil
- }
- }
-
- // idle -> active
- if peer == nil && newPeer != nil {
- self.procWg.Add(1)
-
- poolLogger.Debugf("[%s] active mode", sectionName(sec))
- if !blocksRequestsComplete {
- blocksRequestTime = true
- }
- if !blockHashesRequestsComplete && parentHash != nil {
- blockHashesRequestTime = true
- }
- if !init {
- processC = make(chan *poolNode, blockHashesBatchSize)
- missingC = make(chan *poolNode, blockHashesBatchSize)
- i = 0
- missing = 0
- self.wg.Add(1)
- self.procWg.Add(1)
- depth = len(sec.nodes)
- lastMissing = depth
- // if not run at least once fully, launch iterator
- go func() {
- var node *poolNode
- IT:
- for _, node = range sec.nodes {
- select {
- case processC <- node:
- case <-self.quit:
- break IT
- }
- }
- close(processC)
- self.wg.Done()
- self.procWg.Done()
- }()
- } else {
- poolLogger.Debugf("[%s] restore earlier state", sectionName(sec))
- processC = offC
- }
- }
- // reset quitC to current best peer
- if newPeer != nil {
- quitC = newPeer.quitC
- }
- peer = newPeer
-
- case waiter := <-sec.forkC:
- // this case just blocks the process until section is split at the fork
- <-waiter
- init = false
- done = false
- ready = false
-
- case node, ok := <-processC:
- if !ok && !init {
- // channel closed, first iteration finished
- init = true
- done = true
- processC = make(chan *poolNode, missing)
- poolLogger.DebugDetailf("[%s] section initalised: missing %v/%v/%v", sectionName(sec), missing, lastMissing, depth)
- continue LOOP
- }
- if ready {
- i = 0
- missing = 0
- ready = false
- }
- i++
- // if node has no block
- node.lock.RLock()
- block := node.block
- node.lock.RUnlock()
- if block == nil {
- missing++
- hashes = append(hashes, node.hash)
- if len(hashes) == blockBatchSize {
- poolLogger.Debugf("[%s] request %v missing blocks", sectionName(sec), len(hashes))
- self.requestBlocks(blocksRequests, hashes)
- hashes = nil
- }
- missingC <- node
- } else {
- if i == lastMissing {
- if blockChainC == nil {
- insertChain = true
- } else {
- if parentHash == nil {
- parentHash = block.ParentHash()
- poolLogger.Debugf("[%s] found root block [%s]", sectionName(sec), name(parentHash))
- blockHashesRequestTime = true
- }
- }
- }
- }
- if i == lastMissing && init {
- done = true
- }
-
- case <-blockChainC:
- // closed blockChain channel indicates that the blockpool is reached
- // connected to the blockchain, insert the longest chain of blocks
- poolLogger.Debugf("[%s] reached blockchain", sectionName(sec))
- blockChainC = nil
- // switch off hash requests in case they were on
- blockHashesRequestTime = false
- blockHashesRequestTimer = nil
- blockHashesRequestsComplete = true
- // section root has block
- if len(sec.nodes) > 0 && sec.nodes[len(sec.nodes)-1].block != nil {
- insertChain = true
- }
- continue LOOP
-
- } // select
- } // for
-
- close(sec.offC)
-
- self.wg.Done()
- if peer != nil {
- self.procWg.Done()
- }
- }()
- return
-}
-
-func (self *BlockPool) peerError(peerId string, code int, format string, params ...interface{}) {
- self.peersLock.RLock()
- defer self.peersLock.RUnlock()
- peer, ok := self.peers[peerId]
- if ok {
- peer.peerError(code, format, params...)
- }
-}
-
-func (self *BlockPool) requestBlocks(attempts int, hashes [][]byte) {
- self.wg.Add(1)
- self.procWg.Add(1)
- go func() {
- // distribute block request among known peers
- self.peersLock.Lock()
- defer self.peersLock.Unlock()
- peerCount := len(self.peers)
- // on first attempt use the best peer
- if attempts == 0 {
- poolLogger.Debugf("request %v missing blocks from best peer %s", len(hashes), self.peer.id)
- self.peer.requestBlocks(hashes)
- return
- }
- repetitions := int(math.Min(float64(peerCount), float64(blocksRequestRepetition)))
- i := 0
- indexes := rand.Perm(peerCount)[0:repetitions]
- sort.Ints(indexes)
- poolLogger.Debugf("request %v missing blocks from %v/%v peers: chosen %v", len(hashes), repetitions, peerCount, indexes)
- for _, peer := range self.peers {
- if i == indexes[0] {
- poolLogger.Debugf("request %v missing blocks [%x/%x] from peer %s", len(hashes), hashes[0][:4], hashes[len(hashes)-1][:4], peer.id)
- peer.requestBlocks(hashes)
- indexes = indexes[1:]
- if len(indexes) == 0 {
- break
- }
- }
- i++
- }
- self.wg.Done()
- self.procWg.Done()
- }()
-}
-
-func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) {
- self.peersLock.RLock()
- defer self.peersLock.RUnlock()
- if self.peer != nil && self.peer.id == peerId {
- return self.peer, true
- }
- info, ok := self.peers[peerId]
- if !ok {
- return nil, false
- }
- return info, false
-}
-
-func (self *peerInfo) addSection(hash []byte, section *section) (found *section) {
- self.lock.Lock()
- defer self.lock.Unlock()
- key := string(hash)
- found = self.sections[key]
- poolLogger.DebugDetailf("[%s] section process stored for %s", sectionName(section), self.id)
- self.sections[key] = section
- return
-}
-
-func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) {
- if newPeer != nil {
- newPeer.quitC = make(chan bool)
- poolLogger.DebugDetailf("[%s] activate section processes", newPeer.id)
- var addSections []*section
- for hash, section := range newPeer.sections {
- // split sections get reorganised here
- if string(section.top.hash) != hash {
- addSections = append(addSections, section)
- if entry := self.get([]byte(hash)); entry != nil {
- addSections = append(addSections, entry.section)
- }
- }
- }
- for _, section := range addSections {
- newPeer.sections[string(section.top.hash)] = section
- }
- for hash, section := range newPeer.sections {
- // this will block if section process is waiting for peer lock
- select {
- case <-section.offC:
- poolLogger.DebugDetailf("[%s][%x] section process complete - remove", newPeer.id, hash[:4])
- delete(newPeer.sections, hash)
- case section.controlC <- newPeer:
- poolLogger.DebugDetailf("[%s][%x] activates section [%s]", newPeer.id, hash[:4], sectionName(section))
- }
- }
- newPeer.lock.Lock()
- headSection := newPeer.headSection
- currentBlockHash := newPeer.currentBlockHash
- newPeer.lock.Unlock()
- if headSection == nil {
- poolLogger.DebugDetailf("[%s] head section for [%s] not created, requesting info", newPeer.id, name(currentBlockHash))
- self.requestHeadSection(newPeer)
- } else {
- if entry := self.get(currentBlockHash); entry != nil {
- headSection = entry.section
- }
- poolLogger.DebugDetailf("[%s] activate chain at head section [%s] for current head [%s]", newPeer.id, sectionName(headSection), name(currentBlockHash))
- self.activateChain(headSection, newPeer)
- }
- }
- if oldPeer != nil {
- poolLogger.DebugDetailf("[%s] quit section processes", oldPeer.id)
- close(oldPeer.quitC)
- }
-}
-
-func (self *BlockPool) getParent(sec *section) *section {
- self.chainLock.RLock()
- defer self.chainLock.RUnlock()
- return sec.parent
-}
-
-func (self *BlockPool) getChild(sec *section) *section {
- self.chainLock.RLock()
- defer self.chainLock.RUnlock()
- return sec.child
-}
-
-func newSection() (sec *section) {
- sec = &section{
- controlC: make(chan *peerInfo),
- suicideC: make(chan bool),
- blockChainC: make(chan bool),
- offC: make(chan bool),
- forkC: make(chan chan bool),
- }
- return
-}
-
-// link should only be called under chainLock
-func (self *BlockPool) link(parent *section, child *section) {
- if parent != nil {
- exChild := parent.child
- parent.child = child
- if exChild != nil && exChild != child {
- poolLogger.Debugf("[%s] chain fork [%s] -> [%s]", sectionName(parent), sectionName(exChild), sectionName(child))
- exChild.parent = nil
- }
- }
- if child != nil {
- exParent := child.parent
- if exParent != nil && exParent != parent {
- poolLogger.Debugf("[%s] chain reverse fork [%s] -> [%s]", sectionName(child), sectionName(exParent), sectionName(parent))
- exParent.child = nil
- }
- child.parent = parent
- }
-}
-
-func (self *BlockPool) get(hash []byte) (node *poolEntry) {
- self.lock.RLock()
- defer self.lock.RUnlock()
- return self.pool[string(hash)]
-}
-
-func (self *BlockPool) set(hash []byte, node *poolEntry) {
- self.lock.Lock()
- defer self.lock.Unlock()
- self.pool[string(hash)] = node
-}
generated by cgit (git 2.34.1) at 2024-11-16 15:10:31 +0800