path: root/eth/block_pool.go

package eth

import (
    "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      = 10 // seconds
    blocksRequestRepetition    = 1
    blockHashesRequestInterval = 10 // seconds
    blocksRequestMaxIdleRounds = 10
    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
}

type BlockPool struct {
    lock sync.RWMutex
    pool map[string]*poolNode

    peersLock sync.RWMutex
    peers     map[string]*peerInfo
    peer      *peerInfo

    quit    chan bool
    wg      sync.WaitGroup
    running bool

    // the minimal interface with blockchain
    hasBlock    func(hash []byte) bool
    insertChain func(types.Blocks) error
<<<<<<< HEAD
    verifyPoW   func(pow.Block) bool
=======
    verifyPoW   func(*types.Block) bool
>>>>>>> blockpool rewritten , tests broken FIXME
}

type peerInfo struct {
    lock sync.RWMutex

    td           *big.Int
    currentBlock []byte
    id           string

    requestBlockHashes func([]byte) error
    requestBlocks      func([][]byte) error
    peerError          func(int, string, ...interface{})

    sections map[string]*section
    roots    []*poolNode
    quitC    chan bool
}

<<<<<<< HEAD
func NewBlockPool(hasBlock func(hash []byte) bool, insertChain func(types.Blocks) error, verifyPoW func(pow.Block) bool,
=======
func NewBlockPool(hasBlock func(hash []byte) bool, insertChain func(types.Blocks) error, verifyPoW func(*types.Block) bool,
>>>>>>> blockpool rewritten , tests broken FIXME
) *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.pool = make(map[string]*poolNode)
    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.lock.Lock()
    self.peersLock.Lock()
    self.peers = nil
    self.pool = nil
    self.peer = nil
    self.wg.Wait()
    self.lock.Unlock()
    self.peersLock.Unlock()
    poolLogger.Infoln("Stopped")

}

// 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, 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 := &peerInfo{
        td:                 td,
        currentBlock:       currentBlock,
        id:                 peerId, //peer.Identity().Pubkey()
        requestBlockHashes: requestBlockHashes,
        requestBlocks:      requestBlocks,
        peerError:          peerError,
    }
    self.peers[peerId] = peer
    poolLogger.Debugf("add new peer %v with td %v", peerId, td)
    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)
        self.peer = peer
        return true
    }
    return false
}

// 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 := self.peers[peerId]
    if peer == nil {
        return
    }
    self.peers[peerId] = nil
    poolLogger.Debugf("remove peer %v", peerId[0:4])

    // if current best peer is removed, need find a better one
    if self.peer != nil && peerId == self.peer.id {
        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
            }
        }
        self.peer.stop(peer)
        peer.start(self.peer)
        if newPeer != nil {
            poolLogger.Debugf("peer %v with td %v promoted to best peer", newPeer.id[0:4], newPeer.td)
        } else {
            poolLogger.Warnln("no peers left")
        }
    }
}

// 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) {

    // check if this peer is the best
    peer, best := self.getPeer(peerId)
    if !best {
        return
    }
    // peer is still the best

    var child *poolNode
    var depth int

    // iterate using next (rlp stream lazy decoder) feeding hashesC
    self.wg.Add(1)
    go func() {
        for {
            select {
            case <-self.quit:
                return
            case <-peer.quitC:
                // if the peer is demoted, no more hashes taken
                break
            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 := &section{}
                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 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()
        }
    }()
}

// 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 {
        return
    }
    if node == nil && !self.hasBlock(hash) {
        self.peerError(peerId, ErrUnrequestedBlock, "%x", hash)
        return
    }
    // validate block for PoW
    if !self.verifyPoW(block) {
        self.peerError(peerId, ErrInvalidPoW, "%x", hash)
    }
    node.Lock()
    node.block = block
    node.source = peerId
    node.Unlock()
}

// 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:
            }

            node = bottom.parent
            bottom.sectionRUnlock()
        }
        // remember root for this peer
        peer.addRoot(node)
        self.wg.Done()
    }()
}

// 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(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

    node.sectionLock()
    defer node.sectionUnlock()
    section := &section{controlC: controlC, resetC: resetC}
    node.section = section

    go func() {
        self.wg.Add(1)
        for {
            node.sectionRLock()
            controlC = node.section.controlC
            node.sectionRUnlock()

            if init {
                // missing blocks read from nodeC
                // initialized section
                if depth == 0 {
                    break
                }
                // 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
                }
            }

            // 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
                } 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
                    }
                    same = false
                }
                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
                done = false
                // processC supposed to be empty and never closed so just swap,  no need to allocate
                tempC := processC
                processC = missingC
                missingC = tempC
            }
            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 not root of chain, switch off
                    poolLogger.Debugf("[%x] parent found, hash requests deactivated (after %v total attempts)\n", hash[0:4], blockHashesRequests)
                    blockHashesRequestTimer = nil
                } 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)
                }
            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)

                    alarm = false
                    blocksRequestTimer = nil
                    blockHashesRequestTimer = nil
                    processC = nil
                }
                if !running && r {
                    poolLogger.Debugf("[%x] on", hash[0:4])

                    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)
                    }
                    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
                    }
                    alarm = true
                    processC = missingC
                    if !once {
                        // if not run at least once fully, launch iterator
                        processC = make(chan *poolNode)
                        missingC = make(chan *poolNode)
                        self.foldUp(orignode, processC)
                        once = true
                    }
                }
                total = lastMissing
            case <-resetC:
                once = false
                init = false
                done = false
            case node, ok := <-processC:
                if !ok {
                    // channel closed, first iteration finished
                    init = true
                    once = true
                    continue
                }
                i++
                // if node has no block
                node.RLock()
                block := node.block
                nhash := node.hash
                knownParent := node.knownParent
                node.RUnlock()
                if !init {
                    depth++
                }
                if block == nil {
                    missing++
                    if !init {
                        total++
                    }
                    hashes = append(hashes, nhash)
                    if len(hashes) == blockBatchSize {
                        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)
                    }
                }
            }
        }
        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)

        self.wg.Done()
        node.sectionLock()
        node.section.controlC = nil
        node.sectionUnlock()
        // this signals that controller not available
    }()
    return section

}

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) requestBlockHashes(hash []byte) {
    self.peersLock.Lock()
    defer self.peersLock.Unlock()
    if self.peer != nil {
        self.peer.requestBlockHashes(hash)
    }
}

func (self *BlockPool) requestBlocks(attempts int, hashes [][]byte) {
    // 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 {
        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)]
    sort.Ints(indexes)
    for _, peer := range self.peers {
        if i == indexes[0] {
            peer.requestBlocks(hashes)
            indexes = indexes[1:]
            if len(indexes) == 0 {
                break
            }
        }
        i++
    }
}

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 {
        panic("unknown peer")
    }
    return info, false
}

func (self *peerInfo) addSection(hash []byte, section *section) {
    self.lock.Lock()
    defer self.lock.Unlock()
    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
    self.controlSections(peer, true)
}

//  (re)starts process without requests, only suicide timer
func (self *peerInfo) stop(peer *peerInfo) {
    self.lock.RLock()
    defer self.lock.RUnlock()
    close(self.quitC)
    self.controlSections(peer, false)
}

func (self *peerInfo) controlSections(peer *peerInfo, on bool) {
    if peer != nil {
        peer.lock.RLock()
        defer peer.lock.RUnlock()
    }
    for hash, section := range peer.sections {
        if section.done() {
            delete(self.sections, hash)
        }
        _, exists := peer.sections[hash]
        if on || peer == nil || exists {
            if on {
                // self is best peer
                section.start()
            } else {
                //  (re)starts process without requests, only suicide timer
                section.stop()
            }
        }
    }
}

// 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

    if child == child.section.bottom {
        bottom = true
    }
    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 := &section{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
        parent.child = child
        if !top {
            parent.section.top = parent
            // restart section process so that shorter section is scanned for blocks
            parent.section.reset()
        }
    }

    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 := &section{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
            }
        }
        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() {
    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) {
    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()
}