path: root/eth/block_pool.go

package eth

import (
    "bytes"
    "fmt"
    "math"
    "math/big"
    "sync"
    "time"

    "github.com/ethereum/go-ethereum/core/types"
    "github.com/ethereum/go-ethereum/ethutil"
    "github.com/ethereum/go-ethereum/event"
    ethlogger "github.com/ethereum/go-ethereum/logger"
)

var poolLogger = ethlogger.NewLogger("Blockpool")

const (
    blockHashesBatchSize   = 256
    blockBatchSize         = 64
    blockRequestInterval   = 10 // seconds
    blockRequestRepetition = 1
    cacheTimeout           = 3 // minutes
    blockTimeout           = 5 // minutes
)

type poolNode struct {
    hash                []byte
    block               *types.Block
    child               *poolNode
    parent              *poolNode
    root                *nodePointer
    knownParent         bool
    suicide             chan bool
    peer                string
    source              string
    blockRequestRoot    bool
    blockRequestControl *bool
    blockRequestQuit    *(chan bool)
}

// the minimal interface for chain manager
type chainManager interface {
    KnownBlock(hash []byte) bool
    AddBlock(*types.Block) error
    CheckPoW(*types.Block) bool
}

type BlockPool struct {
    chainManager chainManager
    eventer      event.TypeMux

    // pool     Pool
    lock sync.Mutex
    pool map[string]*poolNode

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

    quit    chan bool
    wg      sync.WaitGroup
    running bool
}

type peerInfo struct {
    td                 *big.Int
    currentBlock       []byte
    id                 string
    requestBlockHashes func([]byte) error
    requestBlocks      func([][]byte) error
    invalidBlock       func(error)
}

type nodePointer struct {
    hash []byte
}

type peerChangeEvent struct {
    *peerInfo
}

func NewBlockPool(chMgr chainManager) *BlockPool {
    return &BlockPool{
        chainManager: chMgr,
        pool:         make(map[string]*poolNode),
        peers:        make(map[string]*peerInfo),
        quit:         make(chan bool),
        running:      true,
    }
}

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()
    poolLogger.Infoln("Stopped")

}

// 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
func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) {
    // subscribe to peerChangeEvent before we check for best peer
    peerChange := self.eventer.Subscribe(peerChangeEvent{})
    defer peerChange.Unsubscribe()
    // check if this peer is the best
    peer, best := self.getPeer(peerId)
    if !best {
        return
    }
    root := &nodePointer{}
    // peer is still the best
    hashes := make(chan []byte)
    var lastPoolNode *poolNode

    // using a for select loop so that peer change (new best peer) can abort the parallel thread that processes hashes of the earlier best peer
    for {
        hash, ok := next()
        if ok {
            hashes <- hash
        } else {
            break
        }
        select {
        case <-self.quit:
            return
        case <-peerChange.Chan():
            // remember where we left off with this peer
            if lastPoolNode != nil {
                root.hash = lastPoolNode.hash
                go self.killChain(lastPoolNode)
            }
        case hash := <-hashes:
            self.lock.Lock()
            defer self.lock.Unlock()
            // check if known block connecting the downloaded chain to our blockchain
            if self.chainManager.KnownBlock(hash) {
                poolLogger.Infof("known block (%x...)\n", hash[0:4])
                if lastPoolNode != nil {
                    lastPoolNode.knownParent = true
                    go self.requestBlocksLoop(lastPoolNode)
                } else {
                    // all hashes known if topmost one is in blockchain
                }
                return
            }
            //
            var currentPoolNode *poolNode
            // check if lastPoolNode has the correct parent node (hash matching),
            // then just assign to currentPoolNode
            if lastPoolNode != nil && lastPoolNode.parent != nil && bytes.Compare(lastPoolNode.parent.hash, hash) == 0 {
                currentPoolNode = lastPoolNode.parent
            } else {
                // otherwise look up in pool
                currentPoolNode = self.pool[string(hash)]
                // if node does not exist, create it and index in the pool
                if currentPoolNode == nil {
                    currentPoolNode = &poolNode{
                        hash: hash,
                    }
                    self.pool[string(hash)] = currentPoolNode
                }
            }
            // set up parent-child nodes (doubly linked list)
            self.link(currentPoolNode, lastPoolNode)
            // ! we trust the node iff
            // (1) node marked as by the same peer or
            // (2) it has a PoW valid block retrieved
            if currentPoolNode.peer == peer.id || currentPoolNode.block != nil {
                // the trusted checkpoint from which we request hashes down to known head
                lastPoolNode = self.pool[string(currentPoolNode.root.hash)]
                break
            }
            currentPoolNode.peer = peer.id
            currentPoolNode.root = root
            lastPoolNode = currentPoolNode
        }
    }
    // lastPoolNode is nil if and only if the node with stored root hash is already cleaned up
    // after valid block insertion, therefore in this case the blockpool active chain is connected to the blockchain, so no need to request further hashes or request blocks
    if lastPoolNode != nil {
        root.hash = lastPoolNode.hash
        peer.requestBlockHashes(lastPoolNode.hash)
        go self.requestBlocksLoop(lastPoolNode)
    }
    return
}

func (self *BlockPool) requestBlocksLoop(node *poolNode) {
    suicide := time.After(blockTimeout * time.Minute)
    requestTimer := time.After(0)
    var controlChan chan bool
    closedChan := make(chan bool)
    quit := make(chan bool)
    close(closedChan)
    requestBlocks := true
    origNode := node
    self.lock.Lock()
    node.blockRequestRoot = true
    b := false
    control := &b
    node.blockRequestControl = control
    node.blockRequestQuit = &quit
    self.lock.Unlock()
    blocks := 0
    self.wg.Add(1)
loop:
    for {
        if requestBlocks {
            controlChan = closedChan
        } else {
            self.lock.Lock()
            if *node.blockRequestControl {
                controlChan = closedChan
                *node.blockRequestControl = false
            }
            self.lock.Unlock()
        }
        select {
        case <-quit:
            break loop
        case <-suicide:
            go self.killChain(origNode)
            break loop

        case <-requestTimer:
            requestBlocks = true

        case <-controlChan:
            controlChan = nil
            // this iteration takes care of requesting blocks only starting from the first node with a missing block (moving target),
            // max up to the next checkpoint (n.blockRequestRoot true)
            nodes := []*poolNode{}
            n := node
            next := node
            self.lock.Lock()
            for n != nil && (n == node || !n.blockRequestRoot) && (requestBlocks || n.block != nil) {
                if n.block != nil {
                    if len(nodes) == 0 {
                        // nil control indicates that node is not needed anymore
                        // block can be inserted to blockchain and deleted if knownParent
                        n.blockRequestControl = nil
                        blocks++
                        next = next.child
                    } else {
                        // this is needed to indicate that when a new chain forks from an existing one
                        // triggering a reorg will ? renew the blockTimeout period ???
                        // if there is a block but control == nil should start fetching blocks, see link function
                        n.blockRequestControl = control
                    }
                } else {
                    nodes = append(nodes, n)
                    n.blockRequestControl = control
                }
                n = n.child
            }
            // if node is connected to the blockchain, we can immediately start inserting
            // blocks to the blockchain and delete nodes
            if node.knownParent {
                go self.insertChainFrom(node)
            }
            if next.blockRequestRoot && next != node {
                // no more missing blocks till the checkpoint, quitting
                poolLogger.Debugf("fetched %v blocks on active chain, batch %v-%v", blocks, origNode, n)
                break loop
            }
            self.lock.Unlock()

            // reset starting node to the first descendant node with missing block
            node = next
            if !requestBlocks {
                continue
            }
            go self.requestBlocks(nodes)
            requestTimer = time.After(blockRequestInterval * time.Second)
        }
    }
    self.wg.Done()
    return
}

func (self *BlockPool) requestBlocks(nodes []*poolNode) {
    // distribute block request among known peers
    self.peersLock.Lock()
    peerCount := len(self.peers)
    poolLogger.Debugf("requesting %v missing blocks from %v peers", len(nodes), peerCount)
    blockHashes := make([][][]byte, peerCount)
    repetitions := int(math.Max(float64(peerCount)/2.0, float64(blockRequestRepetition)))
    for n, node := range nodes {
        for i := 0; i < repetitions; i++ {
            blockHashes[n%peerCount] = append(blockHashes[n%peerCount], node.hash)
            n++
        }
    }
    i := 0
    for _, peer := range self.peers {
        peer.requestBlocks(blockHashes[i])
        i++
    }
    self.peersLock.Unlock()
}

func (self *BlockPool) insertChainFrom(node *poolNode) {
    self.lock.Lock()
    defer self.lock.Unlock()
    for node != nil && node.blockRequestControl == nil {
        err := self.chainManager.AddBlock(node.block)
        if err != nil {
            poolLogger.Debugf("invalid block %v", node.hash)
            poolLogger.Debugf("penalise peers %v (hash), %v (block)", node.peer, node.source)
            // penalise peer in node.source
            go self.killChain(node)
            return
        }
        poolLogger.Debugf("insert block %v into blockchain", node.hash)
        node = node.child
    }
    // if block insertion succeeds, mark the child as knownParent
    // trigger request blocks reorg
    if node != nil {
        node.knownParent = true
        *(node.blockRequestControl) = true
    }
}

// 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, invalidBlock func(error)) bool {
    self.peersLock.Lock()
    defer self.peersLock.Unlock()
    if self.peers[peerId] != nil {
        panic("peer already added")
    }
    info := &peerInfo{
        td:                 td,
        currentBlock:       currentBlock,
        id:                 peerId, //peer.Identity().Pubkey()
        requestBlockHashes: requestBlockHashes,
        requestBlocks:      requestBlocks,
        invalidBlock:       invalidBlock,
    }
    self.peers[peerId] = info
    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 = info
        self.eventer.Post(peerChangeEvent{info})
        poolLogger.Debugf("peer %v promoted to best peer", peerId)
        requestBlockHashes(currentBlock)
        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()
    if self.peers[peerId] != nil {
        panic("peer already removed")
    }
    self.peers[peerId] = nil
    poolLogger.Debugf("remove peer %v", peerId[0:4])

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

func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) {
    self.peersLock.Lock()
    defer self.peersLock.Unlock()
    if self.peer.id == peerId {
        return self.peer, true
    }
    info, ok := self.peers[peerId]
    if !ok {
        panic("unknown peer")
    }
    return info, false
}

// if same peer gave different chain before, this will overwrite it
// if currentPoolNode existed as a non-leaf node the earlier fork is delinked
// if same parent hash is found, we can abort, we do not allow the same peer to change minds about parent of same hash, if errored first time round, will get penalized.
// if lastPoolNode had a different parent the earlier parent (with entire subtree) is delinked, this situation cannot normally arise though
// just in case reset lastPoolNode as non-root (unlikely)

func (self *BlockPool) link(parent, child *poolNode) {
    // reactivate node scheduled for suicide
    if parent.suicide != nil {
        close(parent.suicide)
        parent.suicide = nil
    }
    if parent.child != child {
        orphan := parent.child
        orphan.parent = nil
        go self.killChain(orphan)
        parent.child = child
    }
    if child != nil {
        if child.parent != parent {
            orphan := child.parent
            orphan.child = nil
            go func() {
                // if it is a aberrant reverse fork, zip down to bottom
                for orphan.parent != nil {
                    orphan = orphan.parent
                }
                self.killChain(orphan)
            }()
            child.parent = parent
        }
        child.knownParent = false
    }
}

func (self *BlockPool) killChain(node *poolNode) {
    if node == nil {
        return
    }
    poolLogger.Debugf("suicide scheduled on node %v", node)
    suicide := make(chan bool)
    self.lock.Lock()
    node.suicide = suicide
    self.lock.Unlock()
    timer := time.After(cacheTimeout * time.Minute)
    self.wg.Add(1)
    select {
    case <-self.quit:
    case <-suicide:
        // cancel suicide = close node.suicide to reactivate node
    case <-timer:
        poolLogger.Debugf("suicide on node %v", node)
        self.lock.Lock()
        defer self.lock.Unlock()
        // proceed up via child links until another suicide root found or chain ends
        // abort request blocks loops that start above
        // and delete nodes from pool then quit the suicide process
        okToAbort := node.blockRequestRoot
        for node != nil && (node.suicide == suicide || node.suicide == nil) {
            self.pool[string(node.hash)] = nil
            if okToAbort && node.blockRequestQuit != nil {
                quit := *(node.blockRequestQuit)
                if quit != nil { // not yet closed
                    *(node.blockRequestQuit) = nil
                    close(quit)
                }
            } else {
                okToAbort = true
            }
            node = node.child
        }
    }
    self.wg.Done()
}

// 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) (err error) {
    hash := block.Hash()
    self.lock.Lock()
    defer self.lock.Unlock()
    node, ok := self.pool[string(hash)]
    if !ok && !self.chainManager.KnownBlock(hash) {
        return fmt.Errorf("unrequested block %x", hash)
    }
    if node.block != nil {
        return
    }
    // validate block for PoW
    if !self.chainManager.CheckPoW(block) {
        return fmt.Errorf("invalid pow on block %x", hash)
    }
    node.block = block
    node.source = peerId
    return nil
}