initial commit for eth blockpool + test

pull/205/head
zelig 10 years ago
parent e74f9f27db
commit 3e38f8af23
  1. 514
      eth/block_pool.go
  2. 198
      eth/block_pool_test.go

@ -0,0 +1,514 @@
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
}

@ -0,0 +1,198 @@
package eth
import (
"bytes"
"fmt"
"log"
"os"
"sync"
"testing"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethutil"
ethlogger "github.com/ethereum/go-ethereum/logger"
)
var sys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel))
type testChainManager struct {
knownBlock func(hash []byte) bool
addBlock func(*types.Block) error
checkPoW func(*types.Block) bool
}
func (self *testChainManager) KnownBlock(hash []byte) bool {
if self.knownBlock != nil {
return self.knownBlock(hash)
}
return false
}
func (self *testChainManager) AddBlock(block *types.Block) error {
if self.addBlock != nil {
return self.addBlock(block)
}
return nil
}
func (self *testChainManager) CheckPoW(block *types.Block) bool {
if self.checkPoW != nil {
return self.checkPoW(block)
}
return false
}
func knownBlock(hashes ...[]byte) (f func([]byte) bool) {
f = func(block []byte) bool {
for _, hash := range hashes {
if bytes.Compare(block, hash) == 0 {
return true
}
}
return false
}
return
}
func addBlock(hashes ...[]byte) (f func(*types.Block) error) {
f = func(block *types.Block) error {
for _, hash := range hashes {
if bytes.Compare(block.Hash(), hash) == 0 {
return fmt.Errorf("invalid by test")
}
}
return nil
}
return
}
func checkPoW(hashes ...[]byte) (f func(*types.Block) bool) {
f = func(block *types.Block) bool {
for _, hash := range hashes {
if bytes.Compare(block.Hash(), hash) == 0 {
return false
}
}
return true
}
return
}
func newTestChainManager(knownBlocks [][]byte, invalidBlocks [][]byte, invalidPoW [][]byte) *testChainManager {
return &testChainManager{
knownBlock: knownBlock(knownBlocks...),
addBlock: addBlock(invalidBlocks...),
checkPoW: checkPoW(invalidPoW...),
}
}
type intToHash map[int][]byte
type hashToInt map[string]int
type testHashPool struct {
intToHash
hashToInt
}
func newHash(i int) []byte {
return crypto.Sha3([]byte(string(i)))
}
func newTestBlockPool(knownBlockIndexes []int, invalidBlockIndexes []int, invalidPoWIndexes []int) (hashPool *testHashPool, blockPool *BlockPool) {
hashPool = &testHashPool{make(intToHash), make(hashToInt)}
knownBlocks := hashPool.indexesToHashes(knownBlockIndexes)
invalidBlocks := hashPool.indexesToHashes(invalidBlockIndexes)
invalidPoW := hashPool.indexesToHashes(invalidPoWIndexes)
blockPool = NewBlockPool(newTestChainManager(knownBlocks, invalidBlocks, invalidPoW))
return
}
func (self *testHashPool) indexesToHashes(indexes []int) (hashes [][]byte) {
for _, i := range indexes {
hash, found := self.intToHash[i]
if !found {
hash = newHash(i)
self.intToHash[i] = hash
self.hashToInt[string(hash)] = i
}
hashes = append(hashes, hash)
}
return
}
func (self *testHashPool) hashesToIndexes(hashes [][]byte) (indexes []int) {
for _, hash := range hashes {
i, found := self.hashToInt[string(hash)]
if !found {
i = -1
}
indexes = append(indexes, i)
}
return
}
type protocolChecker struct {
blockHashesRequests []int
blocksRequests [][]int
invalidBlocks []error
hashPool *testHashPool
lock sync.Mutex
}
// -1 is special: not found (a hash never seen)
func (self *protocolChecker) requestBlockHashesCallBack() (requestBlockHashesCallBack func([]byte) error) {
requestBlockHashesCallBack = func(hash []byte) error {
indexes := self.hashPool.hashesToIndexes([][]byte{hash})
self.lock.Lock()
defer self.lock.Unlock()
self.blockHashesRequests = append(self.blockHashesRequests, indexes[0])
return nil
}
return
}
func (self *protocolChecker) requestBlocksCallBack() (requestBlocksCallBack func([][]byte) error) {
requestBlocksCallBack = func(hashes [][]byte) error {
indexes := self.hashPool.hashesToIndexes(hashes)
self.lock.Lock()
defer self.lock.Unlock()
self.blocksRequests = append(self.blocksRequests, indexes)
return nil
}
return
}
func (self *protocolChecker) invalidBlockCallBack() (invalidBlockCallBack func(error)) {
invalidBlockCallBack = func(err error) {
self.invalidBlocks = append(self.invalidBlocks, err)
}
return
}
func TestAddPeer(t *testing.T) {
ethlogger.AddLogSystem(sys)
knownBlockIndexes := []int{0, 1}
invalidBlockIndexes := []int{2, 3}
invalidPoWIndexes := []int{4, 5}
hashPool, blockPool := newTestBlockPool(knownBlockIndexes, invalidBlockIndexes, invalidPoWIndexes)
// TODO:
// hashPool, blockPool, blockChainChecker = newTestBlockPool(knownBlockIndexes, invalidBlockIndexes, invalidPoWIndexes)
peer0 := &protocolChecker{
// blockHashesRequests: make([]int),
// blocksRequests: make([][]int),
// invalidBlocks: make([]error),
hashPool: hashPool,
}
best := blockPool.AddPeer(ethutil.Big1, newHash(100), "0",
peer0.requestBlockHashesCallBack(),
peer0.requestBlocksCallBack(),
peer0.invalidBlockCallBack(),
)
if !best {
t.Errorf("peer not accepted as best")
}
blockPool.Stop()
}
Loading…
Cancel
Save