aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--cmd/ethereum/flags.go6
-rw-r--r--cmd/ethereum/main.go16
-rw-r--r--core/transaction_pool.go54
-rw-r--r--core/types/block.go77
-rw-r--r--eth/backend.go47
-rw-r--r--eth/block_pool.go1407
-rw-r--r--eth/block_pool_test.go926
-rw-r--r--eth/error.go11
-rw-r--r--eth/protocol.go100
-rw-r--r--eth/protocol_test.go300
-rw-r--r--eth/test/README.md27
-rw-r--r--eth/test/bootstrap.sh9
-rwxr-xr-xeth/test/chains/00.chainbin0 -> 9726 bytes
-rwxr-xr-xeth/test/chains/01.chainbin0 -> 13881 bytes
-rwxr-xr-xeth/test/chains/02.chainbin0 -> 14989 bytes
-rwxr-xr-xeth/test/chains/03.chainbin0 -> 18590 bytes
-rwxr-xr-xeth/test/chains/04.chainbin0 -> 20529 bytes
-rw-r--r--eth/test/mine.sh20
-rw-r--r--eth/test/run.sh53
l---------eth/test/tests/00.chain1
-rw-r--r--eth/test/tests/00.sh13
l---------eth/test/tests/01.chain1
-rw-r--r--eth/test/tests/01.sh18
l---------eth/test/tests/02.chain1
-rw-r--r--eth/test/tests/02.sh19
l---------eth/test/tests/03.chain1
-rw-r--r--eth/test/tests/03.sh14
-rw-r--r--eth/test/tests/04.sh17
-rw-r--r--eth/test/tests/05.sh20
-rw-r--r--eth/test/tests/common.js9
-rw-r--r--eth/test/tests/common.sh20
-rw-r--r--p2p/client_identity_test.go2
-rw-r--r--p2p/message.go10
-rw-r--r--p2p/peer.go28
-rw-r--r--p2p/peer_test.go14
-rw-r--r--p2p/protocol.go54
-rw-r--r--p2p/protocol_test.go82
-rw-r--r--p2p/server.go6
-rw-r--r--p2p/server_test.go2
-rw-r--r--rlp/decode.go57
-rw-r--r--rlp/decode_test.go38
41 files changed, 2303 insertions, 1177 deletions
diff --git a/cmd/ethereum/flags.go b/cmd/ethereum/flags.go
index d27b739c3..275fcf248 100644
--- a/cmd/ethereum/flags.go
+++ b/cmd/ethereum/flags.go
@@ -59,6 +59,8 @@ var (
DumpNumber int
VmType int
ImportChain string
+ SHH bool
+ Dial bool
)
// flags specific to cli client
@@ -94,6 +96,8 @@ func Init() {
flag.BoolVar(&StartWebSockets, "ws", false, "start websocket server")
flag.BoolVar(&NonInteractive, "y", false, "non-interactive mode (say yes to confirmations)")
flag.BoolVar(&UseSeed, "seed", true, "seed peers")
+ flag.BoolVar(&SHH, "shh", true, "whisper protocol (on)")
+ flag.BoolVar(&Dial, "dial", true, "dial out connections (on)")
flag.BoolVar(&GenAddr, "genaddr", false, "create a new priv/pub key")
flag.StringVar(&SecretFile, "import", "", "imports the file given (hex or mnemonic formats)")
flag.StringVar(&ExportDir, "export", "", "exports the session keyring to files in the directory given")
@@ -105,7 +109,7 @@ func Init() {
flag.BoolVar(&DiffTool, "difftool", false, "creates output for diff'ing. Sets LogLevel=0")
flag.StringVar(&DiffType, "diff", "all", "sets the level of diff output [vm, all]. Has no effect if difftool=false")
flag.BoolVar(&ShowGenesis, "genesis", false, "Dump the genesis block")
- flag.StringVar(&ImportChain, "chain", "", "Imports fiven chain")
+ flag.StringVar(&ImportChain, "chain", "", "Imports given chain")
flag.BoolVar(&Dump, "dump", false, "output the ethereum state in JSON format. Sub args [number, hash]")
flag.StringVar(&DumpHash, "hash", "", "specify arg in hex")
diff --git a/cmd/ethereum/main.go b/cmd/ethereum/main.go
index b23852282..8b83bbd37 100644
--- a/cmd/ethereum/main.go
+++ b/cmd/ethereum/main.go
@@ -64,10 +64,14 @@ func main() {
NATType: PMPGateway,
PMPGateway: PMPGateway,
KeyRing: KeyRing,
+ Shh: SHH,
+ Dial: Dial,
})
+
if err != nil {
clilogger.Fatalln(err)
}
+
utils.KeyTasks(ethereum.KeyManager(), KeyRing, GenAddr, SecretFile, ExportDir, NonInteractive)
if Dump {
@@ -112,13 +116,6 @@ func main() {
return
}
- // better reworked as cases
- if StartJsConsole {
- InitJsConsole(ethereum)
- } else if len(InputFile) > 0 {
- ExecJsFile(ethereum, InputFile)
- }
-
if StartRpc {
utils.StartRpc(ethereum, RpcPort)
}
@@ -129,6 +126,11 @@ func main() {
utils.StartEthereum(ethereum, UseSeed)
+ if StartJsConsole {
+ InitJsConsole(ethereum)
+ } else if len(InputFile) > 0 {
+ ExecJsFile(ethereum, InputFile)
+ }
// this blocks the thread
ethereum.WaitForShutdown()
}
diff --git a/core/transaction_pool.go b/core/transaction_pool.go
index fa284e52d..ff6c21aa9 100644
--- a/core/transaction_pool.go
+++ b/core/transaction_pool.go
@@ -7,7 +7,6 @@ import (
"github.com/ethereum/go-ethereum/ethutil"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/logger"
- "gopkg.in/fatih/set.v0"
)
var txplogger = logger.NewLogger("TXP")
@@ -38,7 +37,7 @@ type TxPool struct {
quit chan bool
// The actual pool
//pool *list.List
- pool *set.Set
+ txs map[string]*types.Transaction
SecondaryProcessor TxProcessor
@@ -49,21 +48,19 @@ type TxPool struct {
func NewTxPool(eventMux *event.TypeMux) *TxPool {
return &TxPool{
- pool: set.New(),
+ txs: make(map[string]*types.Transaction),
queueChan: make(chan *types.Transaction, txPoolQueueSize),
quit: make(chan bool),
eventMux: eventMux,
}
}
-func (pool *TxPool) addTransaction(tx *types.Transaction) {
- pool.pool.Add(tx)
-
- // Broadcast the transaction to the rest of the peers
- pool.eventMux.Post(TxPreEvent{tx})
-}
-
func (pool *TxPool) ValidateTransaction(tx *types.Transaction) error {
+ hash := tx.Hash()
+ if pool.txs[string(hash)] != nil {
+ return fmt.Errorf("Known transaction (%x)", hash[0:4])
+ }
+
if len(tx.To()) != 0 && len(tx.To()) != 20 {
return fmt.Errorf("Invalid recipient. len = %d", len(tx.To()))
}
@@ -95,18 +92,17 @@ func (pool *TxPool) ValidateTransaction(tx *types.Transaction) error {
return nil
}
-func (self *TxPool) Add(tx *types.Transaction) error {
- hash := tx.Hash()
- if self.pool.Has(tx) {
- return fmt.Errorf("Known transaction (%x)", hash[0:4])
- }
+func (self *TxPool) addTx(tx *types.Transaction) {
+ self.txs[string(tx.Hash())] = tx
+}
+func (self *TxPool) Add(tx *types.Transaction) error {
err := self.ValidateTransaction(tx)
if err != nil {
return err
}
- self.addTransaction(tx)
+ self.addTx(tx)
var to string
if len(tx.To()) > 0 {
@@ -124,7 +120,7 @@ func (self *TxPool) Add(tx *types.Transaction) error {
}
func (self *TxPool) Size() int {
- return self.pool.Size()
+ return len(self.txs)
}
func (self *TxPool) AddTransactions(txs []*types.Transaction) {
@@ -137,43 +133,39 @@ func (self *TxPool) AddTransactions(txs []*types.Transaction) {
}
}
-func (pool *TxPool) GetTransactions() []*types.Transaction {
- txList := make([]*types.Transaction, pool.Size())
+func (self *TxPool) GetTransactions() (txs types.Transactions) {
+ txs = make(types.Transactions, self.Size())
i := 0
- pool.pool.Each(func(v interface{}) bool {
- txList[i] = v.(*types.Transaction)
+ for _, tx := range self.txs {
+ txs[i] = tx
i++
+ }
- return true
- })
-
- return txList
+ return
}
func (pool *TxPool) RemoveInvalid(query StateQuery) {
var removedTxs types.Transactions
- pool.pool.Each(func(v interface{}) bool {
- tx := v.(*types.Transaction)
+ for _, tx := range pool.txs {
sender := query.GetAccount(tx.From())
err := pool.ValidateTransaction(tx)
if err != nil || sender.Nonce >= tx.Nonce() {
removedTxs = append(removedTxs, tx)
}
+ }
- return true
- })
pool.RemoveSet(removedTxs)
}
func (self *TxPool) RemoveSet(txs types.Transactions) {
for _, tx := range txs {
- self.pool.Remove(tx)
+ delete(self.txs, string(tx.Hash()))
}
}
func (pool *TxPool) Flush() []*types.Transaction {
txList := pool.GetTransactions()
- pool.pool.Clear()
+ pool.txs = make(map[string]*types.Transaction)
return txList
}
diff --git a/core/types/block.go b/core/types/block.go
index b59044bfc..7e19d003f 100644
--- a/core/types/block.go
+++ b/core/types/block.go
@@ -67,10 +67,13 @@ func (self *Header) HashNoNonce() []byte {
}
type Block struct {
- header *Header
- uncles []*Header
- transactions Transactions
- Td *big.Int
+ // Preset Hash for mock
+ HeaderHash []byte
+ ParentHeaderHash []byte
+ header *Header
+ uncles []*Header
+ transactions Transactions
+ Td *big.Int
receipts Receipts
Reward *big.Int
@@ -99,41 +102,19 @@ func NewBlockWithHeader(header *Header) *Block {
}
func (self *Block) DecodeRLP(s *rlp.Stream) error {
- if _, err := s.List(); err != nil {
- return err
- }
-
- var header Header
- if err := s.Decode(&header); err != nil {
- return err
- }
-
- var transactions []*Transaction
- if err := s.Decode(&transactions); err != nil {
- return err
+ var extblock struct {
+ Header *Header
+ Txs []*Transaction
+ Uncles []*Header
+ TD *big.Int // optional
}
-
- var uncleHeaders []*Header
- if err := s.Decode(&uncleHeaders); err != nil {
- return err
- }
-
- var tdBytes []byte
- if err := s.Decode(&tdBytes); err != nil {
- // If this block comes from the network that's fine. If loaded from disk it should be there
- // Blocks don't store their Td when propagated over the network
- } else {
- self.Td = ethutil.BigD(tdBytes)
- }
-
- if err := s.ListEnd(); err != nil {
+ if err := s.Decode(&extblock); err != nil {
return err
}
-
- self.header = &header
- self.uncles = uncleHeaders
- self.transactions = transactions
-
+ self.header = extblock.Header
+ self.uncles = extblock.Uncles
+ self.transactions = extblock.Txs
+ self.Td = extblock.TD
return nil
}
@@ -189,23 +170,35 @@ func (self *Block) RlpDataForStorage() interface{} {
// Header accessors (add as you need them)
func (self *Block) Number() *big.Int { return self.header.Number }
func (self *Block) NumberU64() uint64 { return self.header.Number.Uint64() }
-func (self *Block) ParentHash() []byte { return self.header.ParentHash }
func (self *Block) Bloom() []byte { return self.header.Bloom }
func (self *Block) Coinbase() []byte { return self.header.Coinbase }
func (self *Block) Time() int64 { return int64(self.header.Time) }
func (self *Block) GasLimit() *big.Int { return self.header.GasLimit }
func (self *Block) GasUsed() *big.Int { return self.header.GasUsed }
-func (self *Block) Hash() []byte { return self.header.Hash() }
func (self *Block) Trie() *ptrie.Trie { return ptrie.New(self.header.Root, ethutil.Config.Db) }
+func (self *Block) SetRoot(root []byte) { self.header.Root = root }
func (self *Block) State() *state.StateDB { return state.New(self.Trie()) }
func (self *Block) Size() ethutil.StorageSize { return ethutil.StorageSize(len(ethutil.Encode(self))) }
-func (self *Block) SetRoot(root []byte) { self.header.Root = root }
-// Implement block.Pow
+// Implement pow.Block
func (self *Block) Difficulty() *big.Int { return self.header.Difficulty }
func (self *Block) N() []byte { return self.header.Nonce }
-func (self *Block) HashNoNonce() []byte {
- return crypto.Sha3(ethutil.Encode(self.header.rlpData(false)))
+func (self *Block) HashNoNonce() []byte { return self.header.HashNoNonce() }
+
+func (self *Block) Hash() []byte {
+ if self.HeaderHash != nil {
+ return self.HeaderHash
+ } else {
+ return self.header.Hash()
+ }
+}
+
+func (self *Block) ParentHash() []byte {
+ if self.ParentHeaderHash != nil {
+ return self.ParentHeaderHash
+ } else {
+ return self.header.ParentHash
+ }
}
func (self *Block) String() string {
diff --git a/eth/backend.go b/eth/backend.go
index db8e8e029..065a4f7d8 100644
--- a/eth/backend.go
+++ b/eth/backend.go
@@ -36,6 +36,9 @@ type Config struct {
NATType string
PMPGateway string
+ Shh bool
+ Dial bool
+
KeyManager *crypto.KeyManager
}
@@ -130,11 +133,13 @@ func New(config *Config) (*Ethereum, error) {
insertChain := eth.chainManager.InsertChain
eth.blockPool = NewBlockPool(hasBlock, insertChain, ezp.Verify)
- // Start services
- eth.txPool.Start()
-
ethProto := EthProtocol(eth.txPool, eth.chainManager, eth.blockPool)
- protocols := []p2p.Protocol{ethProto, eth.whisper.Protocol()}
+ protocols := []p2p.Protocol{ethProto}
+
+ if config.Shh {
+ eth.whisper = whisper.New()
+ protocols = append(protocols, eth.whisper.Protocol())
+ }
nat, err := p2p.ParseNAT(config.NATType, config.PMPGateway)
if err != nil {
@@ -142,12 +147,16 @@ func New(config *Config) (*Ethereum, error) {
}
eth.net = &p2p.Server{
- Identity: clientId,
- MaxPeers: config.MaxPeers,
- Protocols: protocols,
- ListenAddr: ":" + config.Port,
- Blacklist: eth.blacklist,
- NAT: nat,
+ Identity: clientId,
+ MaxPeers: config.MaxPeers,
+ Protocols: protocols,
+ Blacklist: eth.blacklist,
+ NAT: nat,
+ NoDial: !config.Dial,
+ }
+
+ if len(config.Port) > 0 {
+ eth.net.ListenAddr = ":" + config.Port
}
return eth, nil
@@ -219,8 +228,14 @@ func (s *Ethereum) Start(seed bool) error {
if err != nil {
return err
}
+
+ // Start services
+ s.txPool.Start()
s.blockPool.Start()
- s.whisper.Start()
+
+ if s.whisper != nil {
+ s.whisper.Start()
+ }
// broadcast transactions
s.txSub = s.eventMux.Subscribe(core.TxPreEvent{})
@@ -268,7 +283,9 @@ func (s *Ethereum) Stop() {
s.txPool.Stop()
s.eventMux.Stop()
s.blockPool.Stop()
- s.whisper.Stop()
+ if s.whisper != nil {
+ s.whisper.Stop()
+ }
logger.Infoln("Server stopped")
close(s.shutdownChan)
@@ -285,16 +302,16 @@ func (self *Ethereum) txBroadcastLoop() {
// automatically stops if unsubscribe
for obj := range self.txSub.Chan() {
event := obj.(core.TxPreEvent)
- self.net.Broadcast("eth", TxMsg, []interface{}{event.Tx.RlpData()})
+ self.net.Broadcast("eth", TxMsg, event.Tx.RlpData())
}
}
func (self *Ethereum) blockBroadcastLoop() {
// automatically stops if unsubscribe
- for obj := range self.txSub.Chan() {
+ for obj := range self.blockSub.Chan() {
switch ev := obj.(type) {
case core.NewMinedBlockEvent:
- self.net.Broadcast("eth", NewBlockMsg, ev.Block.RlpData())
+ self.net.Broadcast("eth", NewBlockMsg, ev.Block.RlpData(), ev.Block.Td)
}
}
}
diff --git a/eth/block_pool.go b/eth/block_pool.go
index 7cfbc63f8..519c9fc13 100644
--- a/eth/block_pool.go
+++ b/eth/block_pool.go
@@ -1,6 +1,7 @@
package eth
import (
+ "fmt"
"math"
"math/big"
"math/rand"
@@ -19,37 +20,45 @@ var poolLogger = ethlogger.NewLogger("Blockpool")
const (
blockHashesBatchSize = 256
blockBatchSize = 64
- blocksRequestInterval = 10 // seconds
+ blocksRequestInterval = 500 // ms
blocksRequestRepetition = 1
- blockHashesRequestInterval = 10 // seconds
- blocksRequestMaxIdleRounds = 10
+ blockHashesRequestInterval = 500 // ms
+ blocksRequestMaxIdleRounds = 100
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
+ 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
- pool map[string]*poolNode
+ 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
@@ -70,8 +79,23 @@ type peerInfo struct {
peerError func(int, string, ...interface{})
sections map[string]*section
- roots []*poolNode
- quitC chan bool
+
+ 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,
@@ -92,7 +116,9 @@ func (self *BlockPool) Start() {
}
self.running = true
self.quit = make(chan bool)
- self.pool = make(map[string]*poolNode)
+ self.flushC = make(chan bool)
+ self.pool = make(map[string]*poolEntry)
+
self.lock.Unlock()
self.peersLock.Lock()
@@ -110,50 +136,116 @@ func (self *BlockPool) Stop() {
return
}
self.running = false
+
self.lock.Unlock()
- poolLogger.Infoln("Stopping")
+ poolLogger.Infoln("Stopping...")
close(self.quit)
- self.lock.Lock()
+ self.wg.Wait()
+
self.peersLock.Lock()
self.peers = nil
- self.pool = nil
self.peer = nil
- self.wg.Wait()
- self.lock.Unlock()
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, 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)
+ peer, ok := self.peers[peerId]
+ if ok {
+ poolLogger.Debugf("Update peer %v with td %v and current block %x", peerId, td, currentBlock[:4])
+ peer.td = td
+ peer.currentBlock = currentBlock
+ } else {
+ peer = &peerInfo{
+ td: td,
+ currentBlock: currentBlock,
+ id: peerId, //peer.Identity().Pubkey()
+ requestBlockHashes: requestBlockHashes,
+ requestBlocks: requestBlocks,
+ peerError: peerError,
+ sections: make(map[string]*section),
+ }
+ self.peers[peerId] = peer
+ poolLogger.Debugf("add new peer %v with td %v and current block %x", peerId, td, currentBlock[:4])
+ }
+ // check peer current head
+ if self.hasBlock(currentBlock) {
+ // 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 hashes from %s", peerId, name(currentBlock))
+ peer.requestBlockHashes(currentBlock)
+ return true
+ }
+
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)
+ poolLogger.Debugf("peer %v promoted best peer", peerId)
+ self.switchPeer(self.peer, peer)
self.peer = peer
return true
}
@@ -164,15 +256,15 @@ func (self *BlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string,
func (self *BlockPool) RemovePeer(peerId string) {
self.peersLock.Lock()
defer self.peersLock.Unlock()
- peer := self.peers[peerId]
- if peer == nil {
+ peer, ok := self.peers[peerId]
+ if !ok {
return
}
- self.peers[peerId] = nil
- poolLogger.Debugf("remove peer %v", peerId[0:4])
+ delete(self.peers, peerId)
+ poolLogger.Debugf("remove peer %v", peerId)
// if current best peer is removed, need find a better one
- if self.peer != nil && peerId == self.peer.id {
+ if self.peer == peer {
var newPeer *peerInfo
max := ethutil.Big0
// peer with the highest self-acclaimed TD is chosen
@@ -182,12 +274,12 @@ func (self *BlockPool) RemovePeer(peerId string) {
newPeer = info
}
}
- self.peer.stop(peer)
- peer.start(self.peer)
+ self.peer = newPeer
+ self.switchPeer(peer, newPeer)
if newPeer != nil {
- poolLogger.Debugf("peer %v with td %v promoted to best peer", newPeer.id[0:4], newPeer.td)
+ poolLogger.Debugf("peer %v with td %v promoted to best peer", newPeer.id, newPeer.td)
} else {
- poolLogger.Warnln("no peers left")
+ poolLogger.Warnln("no peers")
}
}
}
@@ -200,97 +292,132 @@ func (self *BlockPool) RemovePeer(peerId string) {
// 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
+ // register with peer manager loop
+
peer, best := self.getPeer(peerId)
if !best {
return
}
// peer is still the best
+ poolLogger.Debugf("adding hashes for best peer %s", peerId)
- var child *poolNode
- var depth int
+ var size, n int
+ var hash []byte
+ var ok bool
+ var section, child, parent *section
+ var entry *poolEntry
+ var nodes []*poolNode
+LOOP:
// 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,
+ for hash, ok = next(); ok; hash, ok = next() {
+ n++
+ select {
+ case <-self.quit:
+ return
+ case <-peer.quitC:
+ // if the peer is demoted, no more hashes taken
+ peer = 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 section != nil {
+ self.connectToBlockChain(section)
+ } else {
+ if child != nil {
+ self.connectToBlockChain(child)
}
- self.set(hash, parent)
- poolLogger.Debugf("create potential block for %x...", hash[0:4])
-
- depth++
- child = parent
}
+ break LOOP
}
- 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()
+ // 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
+ 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 {
+ section = 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(section, nodes)
+ poolLogger.DebugDetailf("[%s]->[%s](%v)->[%s] new chain section", sectionName(parent), sectionName(section), size, sectionName(child))
+ self.link(parent, section)
+ self.link(section, child)
+ } else {
+ poolLogger.DebugDetailf("[%s]->[%s] connecting known sections", sectionName(parent), sectionName(child))
+ self.link(parent, child)
+ }
+
+ self.chainLock.Unlock()
+
+ if parent != nil && peer != nil {
+ self.activateChain(parent, peer)
+ poolLogger.Debugf("[%s] activate parent section [%s]", name(parent.top.hash), sectionName(parent))
+ }
+
+ if section != nil {
+ peer.addSection(section.top.hash, section)
+ section.controlC <- peer
+ poolLogger.Debugf("[%s] activate new section", sectionName(section))
+ }
+}
+
+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
@@ -299,67 +426,114 @@ func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string)
// 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 {
+ if self.hasBlock(hash) {
+ poolLogger.DebugDetailf("block [%s] already known", name(hash))
return
}
- if node == nil && !self.hasBlock(hash) {
+ entry := self.get(hash)
+ if entry == nil {
+ poolLogger.Warnf("unrequested block [%x] by peer %s", hash, peerId)
self.peerError(peerId, ErrUnrequestedBlock, "%x", hash)
return
}
+
+ node := entry.node
+ node.lock.Lock()
+ defer node.lock.Unlock()
+
+ // check if block already present
+ if node.block != nil {
+ poolLogger.DebugDetailf("block [%x] already sent by %s", name(hash), node.blockBy)
+ return
+ }
+
// validate block for PoW
if !self.verifyPoW(block) {
+ poolLogger.Warnf("invalid pow on block [%x] by peer %s", hash, peerId)
self.peerError(peerId, ErrInvalidPoW, "%x", hash)
+ return
}
- node.Lock()
+
+ poolLogger.Debugf("added block [%s] sent by peer %s", name(hash), peerId)
node.block = block
- node.source = peerId
- node.Unlock()
+ node.blockBy = peerId
+
}
-// 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:
- }
+func (self *BlockPool) connectToBlockChain(section *section) {
+ select {
+ case <-section.offC:
+ self.addSectionToBlockChain(section)
+ case <-section.blockChainC:
+ default:
+ close(section.blockChainC)
+ }
+}
- node = bottom.parent
- bottom.sectionRUnlock()
+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
}
- // remember root for this peer
- peer.addRoot(node)
- self.wg.Done()
- }()
+ 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
@@ -370,261 +544,315 @@ func (self *BlockPool) activateChain(node *poolNode, peer *peerInfo, on bool) {
// - 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
+func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
+
+ for i, node := range nodes {
+ entry := &poolEntry{node: node, section: section, index: i}
+ self.set(node.hash, entry)
+ }
+ section.bottom = nodes[len(nodes)-1]
+ section.top = nodes[0]
+ section.nodes = nodes
+ poolLogger.DebugDetailf("[%s] setup section process", sectionName(section))
+
+ self.wg.Add(1)
go func() {
- self.wg.Add(1)
+
+ // absolute time after which sub-chain is killed if not complete (some blocks are missing)
+ suicideTimer := time.After(blockTimeout * time.Minute)
+
+ 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 = section.blockChainC
+
+ LOOP:
for {
- node.sectionRLock()
- controlC = node.section.controlC
- node.sectionRUnlock()
-
- if init {
- // missing blocks read from nodeC
- // initialized section
- if depth == 0 {
- break
+
+ if insertChain {
+ insertChain = false
+ rest, err := self.addSectionToBlockChain(section)
+ if err != nil {
+ close(section.suicideC)
+ continue LOOP
}
- // 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
+ if rest == 0 {
+ blocksRequestsComplete = true
+ child := self.getChild(section)
+ if child != nil {
+ self.connectToBlockChain(child)
+ }
}
}
- // 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
+ 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(section), 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(section), blocksRequests, missing, lastMissing, depth)
+ blocksRequestsComplete = true
+ blocksRequestTimer = nil
+ blocksRequestTime = false
+ } else {
+ // 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(section), idle, blocksRequests, missing, lastMissing, depth)
+ close(section.suicideC)
+ }
+ } else {
+ idle = 0
}
+ same = true
} else {
- idle = 0
+ same = false
}
- 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
+ ready = true
done = false
- // processC supposed to be empty and never closed so just swap, no need to allocate
- tempC := processC
- processC = missingC
- missingC = tempC
+ // save a new processC (blocks still missing)
+ offC = missingC
+ missingC = processC
+ // put processC offline
+ processC = nil
}
- 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 ready && blocksRequestTime && !blocksRequestsComplete {
+ poolLogger.DebugDetailf("[%s] check if new blocks arrived (attempt %v): missing %v/%v/%v", sectionName(section), blocksRequests, missing, lastMissing, depth)
+ blocksRequestTimer = time.After(blocksRequestInterval * time.Millisecond)
+ blocksRequestTime = false
+ processC = offC
+ }
+
+ if blockHashesRequestTime {
+ if self.getParent(section) != 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)
+ poolLogger.DebugDetailf("[%s] parent found, hash requests deactivated (after %v total attempts)\n", sectionName(section), blockHashesRequests)
blockHashesRequestTimer = nil
+ blockHashesRequestsComplete = true
} 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)
+ poolLogger.Debugf("[%s] hash request on root (%v total attempts)\n", sectionName(section), blockHashesRequests)
+ peer.requestBlockHashes(section.bottom.hash)
+ blockHashesRequestTimer = time.After(blockHashesRequestInterval * time.Millisecond)
}
- case r, ok := <-controlC:
- if !ok {
- break
+ blockHashesRequestTime = false
+ }
+
+ select {
+ case <-self.quit:
+ break LOOP
+
+ case <-quitC:
+ // peer quit or demoted, put section in idle mode
+ quitC = nil
+ go func() {
+ section.controlC <- nil
+ }()
+
+ case <-self.purgeC:
+ suicideTimer = time.After(0)
+
+ case <-suicideTimer:
+ close(section.suicideC)
+ poolLogger.Debugf("[%s] timeout. (%v total attempts): missing %v/%v/%v", sectionName(section), blocksRequests, missing, lastMissing, depth)
+
+ case <-section.suicideC:
+ poolLogger.Debugf("[%s] suicide", sectionName(section))
+
+ // first delink from child and parent under chainlock
+ self.chainLock.Lock()
+ self.link(nil, section)
+ self.link(section, nil)
+ self.chainLock.Unlock()
+ // delete node entries from pool index under pool lock
+ self.lock.Lock()
+ for _, node := range section.nodes {
+ delete(self.pool, string(node.hash))
}
- if running && !r {
- poolLogger.Debugf("process on section %x... (%v total attempts): missing %v/%v/%v", hash[0:4], blocksRequests, missing, total, depth)
+ self.lock.Unlock()
+
+ break LOOP
+
+ case <-blocksRequestTimer:
+ poolLogger.DebugDetailf("[%s] block request time", sectionName(section))
+ blocksRequestTime = true
+
+ case <-blockHashesRequestTimer:
+ poolLogger.DebugDetailf("[%s] hash request time", sectionName(section))
+ blockHashesRequestTime = true
+
+ case newPeer = <-section.controlC:
- alarm = false
+ // 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(section), blocksRequests, missing, lastMissing, depth)
+ }
+ blocksRequestTime = false
blocksRequestTimer = nil
+ blockHashesRequestTime = false
blockHashesRequestTimer = nil
- processC = nil
+ if processC != nil {
+ offC = processC
+ 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)
+
+ // idle -> active
+ if peer == nil && newPeer != nil {
+ self.procWg.Add(1)
+
+ poolLogger.Debugf("[%s] active mode", sectionName(section))
+ if !blocksRequestsComplete {
+ blocksRequestTime = true
}
- 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
+ if !blockHashesRequestsComplete {
+ blockHashesRequestTime = true
}
- alarm = true
- processC = missingC
- if !once {
+ 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(section.nodes)
+ lastMissing = depth
// if not run at least once fully, launch iterator
- processC = make(chan *poolNode)
- missingC = make(chan *poolNode)
- self.foldUp(orignode, processC)
- once = true
+ go func() {
+ var node *poolNode
+ IT:
+ for _, node = range section.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(section))
+ processC = offC
}
}
- total = lastMissing
- case <-resetC:
- once = false
+ // reset quitC to current best peer
+ if newPeer != nil {
+ quitC = newPeer.quitC
+ }
+ peer = newPeer
+
+ case waiter := <-section.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 {
+ if !ok && !init {
// channel closed, first iteration finished
init = true
- once = true
- continue
+ done = true
+ processC = make(chan *poolNode, missing)
+ poolLogger.DebugDetailf("[%s] section initalised: missing %v/%v/%v", sectionName(section), missing, lastMissing, depth)
+ continue LOOP
+ }
+ if ready {
+ i = 0
+ missing = 0
+ ready = false
}
i++
// if node has no block
- node.RLock()
+ node.lock.RLock()
block := node.block
- nhash := node.hash
- knownParent := node.knownParent
- node.RUnlock()
- if !init {
- depth++
- }
+ node.lock.RUnlock()
if block == nil {
missing++
- if !init {
- total++
- }
- hashes = append(hashes, nhash)
+ hashes = append(hashes, node.hash)
if len(hashes) == blockBatchSize {
+ poolLogger.Debugf("[%s] request %v missing blocks", sectionName(section), len(hashes))
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)
+ if blockChainC == nil && i == lastMissing {
+ insertChain = true
}
}
- }
- }
- 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)
+ poolLogger.Debugf("[%s] %v/%v/%v/%v", sectionName(section), i, missing, lastMissing, depth)
+ 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(section))
+ blockChainC = nil
+ // switch off hash requests in case they were on
+ blockHashesRequestTime = false
+ blockHashesRequestTimer = nil
+ blockHashesRequestsComplete = true
+ // section root has block
+ if len(section.nodes) > 0 && section.nodes[len(section.nodes)-1].block != nil {
+ insertChain = true
+ }
+ continue LOOP
+
+ } // select
+ } // for
+ poolLogger.Debugf("[%s] section complete: %v block hashes requests - %v block requests - missing %v/%v/%v", sectionName(section), blockHashesRequests, blocksRequests, missing, lastMissing, depth)
+
+ close(section.offC)
self.wg.Done()
- node.sectionLock()
- node.section.controlC = nil
- node.sectionUnlock()
- // this signals that controller not available
+ if peer != nil {
+ self.procWg.Done()
+ }
}()
- return section
-
+ return
}
func (self *BlockPool) peerError(peerId string, code int, format string, params ...interface{}) {
@@ -636,39 +864,39 @@ func (self *BlockPool) peerError(peerId string, code int, format string, 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
+ 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 from peer %s", len(hashes), peer.id)
+ peer.requestBlocks(hashes)
+ indexes = indexes[1:]
+ if len(indexes) == 0 {
+ break
+ }
}
+ i++
}
- i++
- }
+ self.wg.Done()
+ self.procWg.Done()
+ }()
}
func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) {
@@ -679,337 +907,100 @@ func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) {
}
info, ok := self.peers[peerId]
if !ok {
- panic("unknown peer")
+ return nil, false
}
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) {
+func (self *peerInfo) addSection(hash []byte, section *section) (found *section) {
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)
+ key := string(hash)
+ found = self.sections[key]
+ poolLogger.DebugDetailf("[%s] section process %s registered", sectionName(section), self.id)
+ self.sections[key] = section
+ return
+}
+
+func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) {
+ if newPeer != nil {
+ entry := self.get(newPeer.currentBlock)
+ if entry == nil {
+ poolLogger.Debugf("[%s] head block [%s] not found, requesting hashes", newPeer.id, name(newPeer.currentBlock))
+ newPeer.requestBlockHashes(newPeer.currentBlock)
+ } else {
+ poolLogger.Debugf("[%s] head block [%s] found, activate chain at section [%s]", newPeer.id, name(newPeer.currentBlock), sectionName(entry.section))
+ self.activateChain(entry.section, newPeer)
+ }
+ poolLogger.DebugDetailf("[%s] activate section processes", newPeer.id)
+ 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] registered peer with section", newPeer.id, hash[:4])
}
}
- root.sectionRUnlock()
+ newPeer.quitC = make(chan bool)
+ }
+ if oldPeer != nil {
+ close(oldPeer.quitC)
}
- 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 *BlockPool) getParent(sec *section) *section {
+ self.chainLock.RLock()
+ defer self.chainLock.RUnlock()
+ return sec.parent
}
-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()
- }
- }
+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
}
-// 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
+// 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 !top {
- parent.section.top = parent
- // restart section process so that shorter section is scanned for blocks
- parent.section.reset()
+ if exChild != nil && exChild != child {
+ poolLogger.Debugf("[%s] chain fork [%s] -> [%s]", sectionName(parent), sectionName(exChild), sectionName(child))
+ exChild.parent = nil
}
}
-
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
- }
+ 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
- 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() {
+func (self *BlockPool) get(hash []byte) (node *poolEntry) {
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) {
+func (self *BlockPool) set(hash []byte, node *poolEntry) {
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()
-}
diff --git a/eth/block_pool_test.go b/eth/block_pool_test.go
index 315cc748d..b50a314ea 100644
--- a/eth/block_pool_test.go
+++ b/eth/block_pool_test.go
@@ -1,115 +1,65 @@
package eth
import (
- "bytes"
"fmt"
"log"
+ "math/big"
"os"
"sync"
"testing"
+ "time"
"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"
+ "github.com/ethereum/go-ethereum/pow"
)
-var sys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel))
+const waitTimeout = 60 // seconds
-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
-}
+var logsys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugLevel))
-func (self *testChainManager) AddBlock(block *types.Block) error {
- if self.addBlock != nil {
- return self.addBlock(block)
- }
- return nil
-}
+var ini = false
-func (self *testChainManager) CheckPoW(block *types.Block) bool {
- if self.checkPoW != nil {
- return self.checkPoW(block)
+func logInit() {
+ if !ini {
+ ethlogger.AddLogSystem(logsys)
+ ini = true
}
- 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
- }
- }
+// test helpers
+func arrayEq(a, b []int) bool {
+ if len(a) != len(b) {
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
- }
+ for i := range a {
+ if a[i] != b[i] {
+ 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...),
}
+ return true
}
type intToHash map[int][]byte
type hashToInt map[string]int
+// hashPool is a test helper, that allows random hashes to be referred to by integers
type testHashPool struct {
intToHash
hashToInt
+ lock sync.Mutex
}
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) {
+ self.lock.Lock()
+ defer self.lock.Unlock()
for _, i := range indexes {
hash, found := self.intToHash[i]
if !found {
@@ -123,6 +73,8 @@ func (self *testHashPool) indexesToHashes(indexes []int) (hashes [][]byte) {
}
func (self *testHashPool) hashesToIndexes(hashes [][]byte) (indexes []int) {
+ self.lock.Lock()
+ defer self.lock.Unlock()
for _, hash := range hashes {
i, found := self.hashToInt[string(hash)]
if !found {
@@ -133,66 +85,812 @@ func (self *testHashPool) hashesToIndexes(hashes [][]byte) (indexes []int) {
return
}
-type protocolChecker struct {
+// test blockChain is an integer trie
+type blockChain map[int][]int
+
+// blockPoolTester provides the interface between tests and a blockPool
+//
+// refBlockChain is used to guide which blocks will be accepted as valid
+// blockChain gives the current state of the blockchain and
+// accumulates inserts so that we can check the resulting chain
+type blockPoolTester struct {
+ hashPool *testHashPool
+ lock sync.RWMutex
+ refBlockChain blockChain
+ blockChain blockChain
+ blockPool *BlockPool
+ t *testing.T
+}
+
+func newTestBlockPool(t *testing.T) (hashPool *testHashPool, blockPool *BlockPool, b *blockPoolTester) {
+ hashPool = &testHashPool{intToHash: make(intToHash), hashToInt: make(hashToInt)}
+ b = &blockPoolTester{
+ t: t,
+ hashPool: hashPool,
+ blockChain: make(blockChain),
+ refBlockChain: make(blockChain),
+ }
+ b.blockPool = NewBlockPool(b.hasBlock, b.insertChain, b.verifyPoW)
+ blockPool = b.blockPool
+ return
+}
+
+func (self *blockPoolTester) Errorf(format string, params ...interface{}) {
+ fmt.Printf(format+"\n", params...)
+ self.t.Errorf(format, params...)
+}
+
+// blockPoolTester implements the 3 callbacks needed by the blockPool:
+// hasBlock, insetChain, verifyPoW
+func (self *blockPoolTester) hasBlock(block []byte) (ok bool) {
+ self.lock.RLock()
+ defer self.lock.RUnlock()
+ indexes := self.hashPool.hashesToIndexes([][]byte{block})
+ i := indexes[0]
+ _, ok = self.blockChain[i]
+ fmt.Printf("has block %v (%x...): %v\n", i, block[0:4], ok)
+ return
+}
+
+func (self *blockPoolTester) insertChain(blocks types.Blocks) error {
+ self.lock.RLock()
+ defer self.lock.RUnlock()
+ var parent, child int
+ var children, refChildren []int
+ var ok bool
+ for _, block := range blocks {
+ child = self.hashPool.hashesToIndexes([][]byte{block.Hash()})[0]
+ _, ok = self.blockChain[child]
+ if ok {
+ fmt.Printf("block %v already in blockchain\n", child)
+ continue // already in chain
+ }
+ parent = self.hashPool.hashesToIndexes([][]byte{block.ParentHeaderHash})[0]
+ children, ok = self.blockChain[parent]
+ if !ok {
+ return fmt.Errorf("parent %v not in blockchain ", parent)
+ }
+ ok = false
+ var found bool
+ refChildren, found = self.refBlockChain[parent]
+ if found {
+ for _, c := range refChildren {
+ if c == child {
+ ok = true
+ }
+ }
+ if !ok {
+ return fmt.Errorf("invalid block %v", child)
+ }
+ } else {
+ ok = true
+ }
+ if ok {
+ // accept any blocks if parent not in refBlockChain
+ fmt.Errorf("blockchain insert %v -> %v\n", parent, child)
+ self.blockChain[parent] = append(children, child)
+ self.blockChain[child] = nil
+ }
+ }
+ return nil
+}
+
+func (self *blockPoolTester) verifyPoW(pblock pow.Block) bool {
+ return true
+}
+
+// test helper that compares the resulting blockChain to the desired blockChain
+func (self *blockPoolTester) checkBlockChain(blockChain map[int][]int) {
+ for k, v := range self.blockChain {
+ fmt.Printf("got: %v -> %v\n", k, v)
+ }
+ for k, v := range blockChain {
+ fmt.Printf("expected: %v -> %v\n", k, v)
+ }
+ if len(blockChain) != len(self.blockChain) {
+ self.Errorf("blockchain incorrect (zlength differ)")
+ }
+ for k, v := range blockChain {
+ vv, ok := self.blockChain[k]
+ if !ok || !arrayEq(v, vv) {
+ self.Errorf("blockchain incorrect on %v -> %v (!= %v)", k, vv, v)
+ }
+ }
+}
+
+//
+
+// peerTester provides the peer callbacks for the blockPool
+// it registers actual callbacks so that result can be compared to desired behaviour
+// provides helper functions to mock the protocol calls to the blockPool
+type peerTester struct {
blockHashesRequests []int
blocksRequests [][]int
- invalidBlocks []error
+ blocksRequestsMap map[int]bool
+ peerErrors []int
+ blockPool *BlockPool
hashPool *testHashPool
- lock sync.Mutex
+ lock sync.RWMutex
+ id string
+ td int
+ currentBlock int
+ t *testing.T
}
-// -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
+// peerTester constructor takes hashPool and blockPool from the blockPoolTester
+func (self *blockPoolTester) newPeer(id string, td int, cb int) *peerTester {
+ return &peerTester{
+ id: id,
+ td: td,
+ currentBlock: cb,
+ hashPool: self.hashPool,
+ blockPool: self.blockPool,
+ t: self.t,
+ blocksRequestsMap: make(map[int]bool),
}
- 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
+func (self *peerTester) Errorf(format string, params ...interface{}) {
+ fmt.Printf(format+"\n", params...)
+ self.t.Errorf(format, params...)
+}
+
+// helper to compare actual and expected block requests
+func (self *peerTester) checkBlocksRequests(blocksRequests ...[]int) {
+ if len(blocksRequests) > len(self.blocksRequests) {
+ self.Errorf("blocks requests incorrect (length differ)\ngot %v\nexpected %v", self.blocksRequests, blocksRequests)
+ } else {
+ for i, rr := range blocksRequests {
+ r := self.blocksRequests[i]
+ if !arrayEq(r, rr) {
+ self.Errorf("blocks requests incorrect\ngot %v\nexpected %v", self.blocksRequests, blocksRequests)
+ }
+ }
}
- return
}
-func (self *protocolChecker) invalidBlockCallBack() (invalidBlockCallBack func(error)) {
- invalidBlockCallBack = func(err error) {
- self.invalidBlocks = append(self.invalidBlocks, err)
+// helper to compare actual and expected block hash requests
+func (self *peerTester) checkBlockHashesRequests(blocksHashesRequests ...int) {
+ rr := blocksHashesRequests
+ self.lock.RLock()
+ r := self.blockHashesRequests
+ self.lock.RUnlock()
+ if len(r) != len(rr) {
+ self.Errorf("block hashes requests incorrect (length differ)\ngot %v\nexpected %v", r, rr)
+ } else {
+ if !arrayEq(r, rr) {
+ self.Errorf("block hashes requests incorrect\ngot %v\nexpected %v", r, rr)
+ }
+ }
+}
+
+// waiter function used by peer.AddBlocks
+// blocking until requests appear
+// since block requests are sent to any random peers
+// block request map is shared between peers
+// times out after a period
+func (self *peerTester) waitBlocksRequests(blocksRequest ...int) {
+ timeout := time.After(waitTimeout * time.Second)
+ rr := blocksRequest
+ for {
+ self.lock.RLock()
+ r := self.blocksRequestsMap
+ fmt.Printf("[%s] blocks request check %v (%v)\n", self.id, rr, r)
+ i := 0
+ for i = 0; i < len(rr); i++ {
+ _, ok := r[rr[i]]
+ if !ok {
+ break
+ }
+ }
+ self.lock.RUnlock()
+
+ if i == len(rr) {
+ return
+ }
+ time.Sleep(100 * time.Millisecond)
+ select {
+ case <-timeout:
+ default:
+ }
}
- return
}
+// waiter function used by peer.AddBlockHashes
+// blocking until requests appear
+// times out after a period
+func (self *peerTester) waitBlockHashesRequests(blocksHashesRequest int) {
+ timeout := time.After(waitTimeout * time.Second)
+ rr := blocksHashesRequest
+ for i := 0; ; {
+ self.lock.RLock()
+ r := self.blockHashesRequests
+ self.lock.RUnlock()
+ fmt.Printf("[%s] block hash request check %v (%v)\n", self.id, rr, r)
+ for ; i < len(r); i++ {
+ if rr == r[i] {
+ return
+ }
+ }
+ time.Sleep(100 * time.Millisecond)
+ select {
+ case <-timeout:
+ default:
+ }
+ }
+}
+
+// mocks a simple blockchain 0 (genesis) ... n (head)
+func (self *blockPoolTester) initRefBlockChain(n int) {
+ for i := 0; i < n; i++ {
+ self.refBlockChain[i] = []int{i + 1}
+ }
+}
+
+// peerTester functions that mimic protocol calls to the blockpool
+// registers the peer with the blockPool
+func (self *peerTester) AddPeer() bool {
+ hash := self.hashPool.indexesToHashes([]int{self.currentBlock})[0]
+ return self.blockPool.AddPeer(big.NewInt(int64(self.td)), hash, self.id, self.requestBlockHashes, self.requestBlocks, self.peerError)
+}
+
+// peer sends blockhashes if and when gets a request
+func (self *peerTester) AddBlockHashes(indexes ...int) {
+ i := 0
+ fmt.Printf("ready to add block hashes %v\n", indexes)
+
+ self.waitBlockHashesRequests(indexes[0])
+ fmt.Printf("adding block hashes %v\n", indexes)
+ hashes := self.hashPool.indexesToHashes(indexes)
+ next := func() (hash []byte, ok bool) {
+ if i < len(hashes) {
+ hash = hashes[i]
+ ok = true
+ i++
+ }
+ return
+ }
+ self.blockPool.AddBlockHashes(next, self.id)
+}
+
+// peer sends blocks if and when there is a request
+// (in the shared request store, not necessarily to a person)
+func (self *peerTester) AddBlocks(indexes ...int) {
+ hashes := self.hashPool.indexesToHashes(indexes)
+ fmt.Printf("ready to add blocks %v\n", indexes[1:])
+ self.waitBlocksRequests(indexes[1:]...)
+ fmt.Printf("adding blocks %v \n", indexes[1:])
+ for i := 1; i < len(hashes); i++ {
+ fmt.Printf("adding block %v %x\n", indexes[i], hashes[i][:4])
+ self.blockPool.AddBlock(&types.Block{HeaderHash: ethutil.Bytes(hashes[i]), ParentHeaderHash: ethutil.Bytes(hashes[i-1])}, self.id)
+ }
+}
+
+// peer callbacks
+// -1 is special: not found (a hash never seen)
+// records block hashes requests by the blockPool
+func (self *peerTester) requestBlockHashes(hash []byte) error {
+ indexes := self.hashPool.hashesToIndexes([][]byte{hash})
+ fmt.Printf("[%s] blocks hash request %v %x\n", self.id, indexes[0], hash[:4])
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ self.blockHashesRequests = append(self.blockHashesRequests, indexes[0])
+ return nil
+}
+
+// records block requests by the blockPool
+func (self *peerTester) requestBlocks(hashes [][]byte) error {
+ indexes := self.hashPool.hashesToIndexes(hashes)
+ fmt.Printf("blocks request %v %x...\n", indexes, hashes[0][:4])
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ self.blocksRequests = append(self.blocksRequests, indexes)
+ for _, i := range indexes {
+ self.blocksRequestsMap[i] = true
+ }
+ return nil
+}
+
+// records the error codes of all the peerErrors found the blockPool
+func (self *peerTester) peerError(code int, format string, params ...interface{}) {
+ self.peerErrors = append(self.peerErrors, code)
+}
+
+// the actual tests
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(),
- )
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ peer0 := blockPoolTester.newPeer("peer0", 1, 0)
+ peer1 := blockPoolTester.newPeer("peer1", 2, 1)
+ peer2 := blockPoolTester.newPeer("peer2", 3, 2)
+ var peer *peerInfo
+
+ blockPool.Start()
+
+ // pool
+ best := peer0.AddPeer()
+ if !best {
+ t.Errorf("peer0 (TD=1) not accepted as best")
+ }
+ if blockPool.peer.id != "peer0" {
+ t.Errorf("peer0 (TD=1) not set as best")
+ }
+ peer0.checkBlockHashesRequests(0)
+
+ best = peer2.AddPeer()
+ if !best {
+ t.Errorf("peer2 (TD=3) not accepted as best")
+ }
+ if blockPool.peer.id != "peer2" {
+ t.Errorf("peer2 (TD=3) not set as best")
+ }
+ peer2.checkBlockHashesRequests(2)
+
+ best = peer1.AddPeer()
+ if best {
+ t.Errorf("peer1 (TD=2) accepted as best")
+ }
+ if blockPool.peer.id != "peer2" {
+ t.Errorf("peer2 (TD=3) not set any more as best")
+ }
+ if blockPool.peer.td.Cmp(big.NewInt(int64(3))) != 0 {
+ t.Errorf("peer1 TD not set")
+ }
+
+ peer2.td = 4
+ peer2.currentBlock = 3
+ best = peer2.AddPeer()
if !best {
- t.Errorf("peer not accepted as best")
+ t.Errorf("peer2 (TD=4) not accepted as best")
}
+ if blockPool.peer.id != "peer2" {
+ t.Errorf("peer2 (TD=4) not set as best")
+ }
+ if blockPool.peer.td.Cmp(big.NewInt(int64(4))) != 0 {
+ t.Errorf("peer2 TD not updated")
+ }
+ peer2.checkBlockHashesRequests(2, 3)
+
+ peer1.td = 3
+ peer1.currentBlock = 2
+ best = peer1.AddPeer()
+ if best {
+ t.Errorf("peer1 (TD=3) should not be set as best")
+ }
+ if blockPool.peer.id == "peer1" {
+ t.Errorf("peer1 (TD=3) should not be set as best")
+ }
+ peer, best = blockPool.getPeer("peer1")
+ if peer.td.Cmp(big.NewInt(int64(3))) != 0 {
+ t.Errorf("peer1 TD should be updated")
+ }
+
+ blockPool.RemovePeer("peer2")
+ peer, best = blockPool.getPeer("peer2")
+ if peer != nil {
+ t.Errorf("peer2 not removed")
+ }
+
+ if blockPool.peer.id != "peer1" {
+ t.Errorf("existing peer1 (TD=3) should be set as best peer")
+ }
+ peer1.checkBlockHashesRequests(2)
+
+ blockPool.RemovePeer("peer1")
+ peer, best = blockPool.getPeer("peer1")
+ if peer != nil {
+ t.Errorf("peer1 not removed")
+ }
+
+ if blockPool.peer.id != "peer0" {
+ t.Errorf("existing peer0 (TD=1) should be set as best peer")
+ }
+
+ blockPool.RemovePeer("peer0")
+ peer, best = blockPool.getPeer("peer0")
+ if peer != nil {
+ t.Errorf("peer1 not removed")
+ }
+
+ // adding back earlier peer ok
+ peer0.currentBlock = 3
+ best = peer0.AddPeer()
+ if !best {
+ t.Errorf("peer0 (TD=1) should be set as best")
+ }
+
+ if blockPool.peer.id != "peer0" {
+ t.Errorf("peer0 (TD=1) should be set as best")
+ }
+ peer0.checkBlockHashesRequests(0, 0, 3)
+
+ blockPool.Stop()
+
+}
+
+func TestPeerWithKnownBlock(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.refBlockChain[0] = nil
+ blockPoolTester.blockChain[0] = nil
+ // hashPool, blockPool, blockPoolTester := newTestBlockPool()
+ blockPool.Start()
+
+ peer0 := blockPoolTester.newPeer("0", 1, 0)
+ peer0.AddPeer()
+
+ blockPool.Stop()
+ // no request on known block
+ peer0.checkBlockHashesRequests()
+}
+
+func TestSimpleChain(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(2)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 2)
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(2, 1, 0)
+ peer1.AddBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[2] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestInvalidBlock(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(2)
+ blockPoolTester.refBlockChain[2] = []int{}
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 3)
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(3, 2, 1, 0)
+ peer1.AddBlocks(0, 1, 2, 3)
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[2] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+ if len(peer1.peerErrors) == 1 {
+ if peer1.peerErrors[0] != ErrInvalidBlock {
+ t.Errorf("wrong error, got %v, expected %v", peer1.peerErrors[0], ErrInvalidBlock)
+ }
+ } else {
+ t.Errorf("expected invalid block error, got nothing")
+ }
+}
+
+func TestVerifyPoW(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(3)
+ first := false
+ blockPoolTester.blockPool.verifyPoW = func(b pow.Block) bool {
+ bb, _ := b.(*types.Block)
+ indexes := blockPoolTester.hashPool.hashesToIndexes([][]byte{bb.Hash()})
+ if indexes[0] == 1 && !first {
+ first = true
+ return false
+ } else {
+ return true
+ }
+
+ }
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 2)
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(2, 1, 0)
+ peer1.AddBlocks(0, 1, 2)
+ peer1.AddBlocks(0, 1)
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[2] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+ if len(peer1.peerErrors) == 1 {
+ if peer1.peerErrors[0] != ErrInvalidPoW {
+ t.Errorf("wrong error, got %v, expected %v", peer1.peerErrors[0], ErrInvalidPoW)
+ }
+ } else {
+ t.Errorf("expected invalid pow error, got nothing")
+ }
+}
+
+func TestMultiSectionChain(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(5)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 5)
+
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(5, 4, 3)
+ go peer1.AddBlocks(2, 3, 4, 5)
+ go peer1.AddBlockHashes(3, 2, 1, 0)
+ peer1.AddBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[5] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestNewBlocksOnPartialChain(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(7)
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 5)
+
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(5, 4, 3)
+ peer1.AddBlocks(2, 3) // partially complete section
+ // peer1 found new blocks
+ peer1.td = 2
+ peer1.currentBlock = 7
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(7, 6, 5)
+ go peer1.AddBlocks(3, 4, 5, 6, 7)
+ go peer1.AddBlockHashes(3, 2, 1, 0) // tests that hash request from known chain root is remembered
+ peer1.AddBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[7] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestPeerSwitch(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(6)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 5)
+ peer2 := blockPoolTester.newPeer("peer2", 2, 6)
+ peer2.blocksRequestsMap = peer1.blocksRequestsMap
+
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(5, 4, 3)
+ peer1.AddBlocks(2, 3) // section partially complete, block 3 will be preserved after peer demoted
+ peer2.AddPeer() // peer2 is promoted as best peer, peer1 is demoted
+ go peer2.AddBlockHashes(6, 5) //
+ go peer2.AddBlocks(4, 5, 6) // tests that block request for earlier section is remembered
+ go peer1.AddBlocks(3, 4) // tests that connecting section by demoted peer is remembered and blocks are accepted from demoted peer
+ go peer2.AddBlockHashes(3, 2, 1, 0) // tests that known chain section is activated, hash requests from 3 is remembered
+ peer2.AddBlocks(0, 1, 2) // final blocks linking to blockchain sent
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[6] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestPeerDownSwitch(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(6)
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 4)
+ peer2 := blockPoolTester.newPeer("peer2", 2, 6)
+ peer2.blocksRequestsMap = peer1.blocksRequestsMap
+
+ peer2.AddPeer()
+ go peer2.AddBlockHashes(6, 5, 4)
+ peer2.AddBlocks(5, 6) // partially complete, section will be preserved
+ blockPool.RemovePeer("peer2") // peer2 disconnects
+ peer1.AddPeer() // inferior peer1 is promoted as best peer
+ go peer1.AddBlockHashes(4, 3, 2, 1, 0) //
+ go peer1.AddBlocks(3, 4, 5) // tests that section set by demoted peer is remembered and blocks are accepted
+ peer1.AddBlocks(0, 1, 2, 3)
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[6] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestPeerSwitchBack(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(8)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 2, 11)
+ peer2 := blockPoolTester.newPeer("peer2", 1, 8)
+ peer2.blocksRequestsMap = peer1.blocksRequestsMap
+
+ peer2.AddPeer()
+ go peer2.AddBlockHashes(8, 7, 6)
+ go peer2.AddBlockHashes(6, 5, 4)
+ peer2.AddBlocks(4, 5) // section partially complete
+ peer1.AddPeer() // peer1 is promoted as best peer
+ go peer1.AddBlockHashes(11, 10) // only gives useless results
+ blockPool.RemovePeer("peer1") // peer1 disconnects
+ go peer2.AddBlockHashes(4, 3, 2, 1, 0) // tests that asking for hashes from 4 is remembered
+ go peer2.AddBlocks(3, 4, 5, 6, 7, 8) // tests that section 4, 5, 6 and 7, 8 are remembered for missing blocks
+ peer2.AddBlocks(0, 1, 2, 3)
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[8] = []int{}
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+}
+
+func TestForkSimple(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(9)
+ blockPoolTester.refBlockChain[3] = []int{4, 7}
+ delete(blockPoolTester.refBlockChain, 6)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 9)
+ peer2 := blockPoolTester.newPeer("peer2", 2, 6)
+ peer2.blocksRequestsMap = peer1.blocksRequestsMap
+
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(9, 8, 7, 3, 2)
+ peer1.AddBlocks(1, 2, 3, 7, 8, 9)
+ peer2.AddPeer() // peer2 is promoted as best peer
+ go peer2.AddBlockHashes(6, 5, 4, 3, 2) // fork on 3 -> 4 (earlier child: 7)
+ go peer2.AddBlocks(1, 2, 3, 4, 5, 6)
+ go peer2.AddBlockHashes(2, 1, 0)
+ peer2.AddBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[6] = []int{}
+ blockPoolTester.refBlockChain[3] = []int{4}
+ delete(blockPoolTester.refBlockChain, 7)
+ delete(blockPoolTester.refBlockChain, 8)
+ delete(blockPoolTester.refBlockChain, 9)
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+
+}
+
+func TestForkSwitchBackByNewBlocks(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(11)
+ blockPoolTester.refBlockChain[3] = []int{4, 7}
+ delete(blockPoolTester.refBlockChain, 6)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 9)
+ peer2 := blockPoolTester.newPeer("peer2", 2, 6)
+ peer2.blocksRequestsMap = peer1.blocksRequestsMap
+
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(9, 8, 7, 3, 2)
+ peer1.AddBlocks(8, 9) // partial section
+ peer2.AddPeer() //
+ go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
+ peer2.AddBlocks(1, 2, 3, 4, 5, 6) //
+
+ // peer1 finds new blocks
+ peer1.td = 3
+ peer1.currentBlock = 11
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(11, 10, 9)
+ peer1.AddBlocks(7, 8, 9, 10, 11)
+ go peer1.AddBlockHashes(7, 3) // tests that hash request from fork root is remembered
+ go peer1.AddBlocks(3, 7) // tests that block requests on earlier fork are remembered
+ // go peer1.AddBlockHashes(1, 0) // tests that hash request from root of connecting chain section (added by demoted peer) is remembered
+ go peer1.AddBlockHashes(2, 1, 0) // tests that hash request from root of connecting chain section (added by demoted peer) is remembered
+ peer1.AddBlocks(0, 1, 2, 3)
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[11] = []int{}
+ blockPoolTester.refBlockChain[3] = []int{7}
+ delete(blockPoolTester.refBlockChain, 6)
+ delete(blockPoolTester.refBlockChain, 5)
+ delete(blockPoolTester.refBlockChain, 4)
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+
+}
+
+func TestForkSwitchBackByPeerSwitchBack(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(9)
+ blockPoolTester.refBlockChain[3] = []int{4, 7}
+ delete(blockPoolTester.refBlockChain, 6)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 9)
+ peer2 := blockPoolTester.newPeer("peer2", 2, 6)
+ peer2.blocksRequestsMap = peer1.blocksRequestsMap
+
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(9, 8, 7, 3, 2)
+ peer1.AddBlocks(8, 9)
+ peer2.AddPeer() //
+ go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
+ peer2.AddBlocks(2, 3, 4, 5, 6) //
+ blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer
+ peer1.AddBlockHashes(7, 3) // tests that hash request from fork root is remembered
+ go peer1.AddBlocks(3, 7, 8) // tests that block requests on earlier fork are remembered
+ go peer1.AddBlockHashes(2, 1, 0) //
+ peer1.AddBlocks(0, 1, 2, 3)
+
+ blockPool.Wait(waitTimeout * time.Second)
+ blockPool.Stop()
+ blockPoolTester.refBlockChain[9] = []int{}
+ blockPoolTester.refBlockChain[3] = []int{7}
+ delete(blockPoolTester.refBlockChain, 6)
+ delete(blockPoolTester.refBlockChain, 5)
+ delete(blockPoolTester.refBlockChain, 4)
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
+
+}
+
+func TestForkCompleteSectionSwitchBackByPeerSwitchBack(t *testing.T) {
+ logInit()
+ _, blockPool, blockPoolTester := newTestBlockPool(t)
+ blockPoolTester.blockChain[0] = nil
+ blockPoolTester.initRefBlockChain(9)
+ blockPoolTester.refBlockChain[3] = []int{4, 7}
+ delete(blockPoolTester.refBlockChain, 6)
+
+ blockPool.Start()
+
+ peer1 := blockPoolTester.newPeer("peer1", 1, 9)
+ peer2 := blockPoolTester.newPeer("peer2", 2, 6)
+ peer2.blocksRequestsMap = peer1.blocksRequestsMap
+
+ peer1.AddPeer()
+ go peer1.AddBlockHashes(9, 8, 7)
+ peer1.AddBlocks(3, 7, 8, 9) // make sure this section is complete
+ time.Sleep(1 * time.Second)
+ go peer1.AddBlockHashes(7, 3, 2) // block 3/7 is section boundary
+ peer1.AddBlocks(2, 3) // partially complete sections
+ peer2.AddPeer() //
+ go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
+ peer2.AddBlocks(2, 3, 4, 5, 6) // block 2 still missing.
+ blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer
+ peer1.AddBlockHashes(7, 3) // tests that hash request from fork root is remembered even though section process completed
+ go peer1.AddBlockHashes(2, 1, 0) //
+ peer1.AddBlocks(0, 1, 2)
+
+ blockPool.Wait(waitTimeout * time.Second)
blockPool.Stop()
+ blockPoolTester.refBlockChain[9] = []int{}
+ blockPoolTester.refBlockChain[3] = []int{7}
+ delete(blockPoolTester.refBlockChain, 6)
+ delete(blockPoolTester.refBlockChain, 5)
+ delete(blockPoolTester.refBlockChain, 4)
+ blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
diff --git a/eth/error.go b/eth/error.go
index d1daad575..1d9f80638 100644
--- a/eth/error.go
+++ b/eth/error.go
@@ -52,18 +52,17 @@ func ProtocolError(code int, format string, params ...interface{}) (err *protoco
}
func (self protocolError) Error() (message string) {
- message = self.message
- if message == "" {
- message, ok := errorToString[self.Code]
+ if len(message) == 0 {
+ var ok bool
+ self.message, ok = errorToString[self.Code]
if !ok {
panic("invalid error code")
}
if self.format != "" {
- message += ": " + fmt.Sprintf(self.format, self.params...)
+ self.message += ": " + fmt.Sprintf(self.format, self.params...)
}
- self.message = message
}
- return
+ return self.message
}
func (self *protocolError) Fatal() bool {
diff --git a/eth/protocol.go b/eth/protocol.go
index 7c5d09489..b67e5aaea 100644
--- a/eth/protocol.go
+++ b/eth/protocol.go
@@ -3,7 +3,7 @@ package eth
import (
"bytes"
"fmt"
- "math"
+ "io"
"math/big"
"github.com/ethereum/go-ethereum/core/types"
@@ -95,14 +95,13 @@ func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPoo
blockPool: blockPool,
rw: rw,
peer: peer,
- id: (string)(peer.Identity().Pubkey()),
+ id: fmt.Sprintf("%x", peer.Identity().Pubkey()[:8]),
}
err = self.handleStatus()
if err == nil {
for {
err = self.handle()
if err != nil {
- fmt.Println(err)
self.blockPool.RemovePeer(self.id)
break
}
@@ -117,7 +116,7 @@ func (self *ethProtocol) handle() error {
return err
}
if msg.Size > ProtocolMaxMsgSize {
- return ProtocolError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
+ return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
}
// make sure that the payload has been fully consumed
defer msg.Discard()
@@ -125,76 +124,87 @@ func (self *ethProtocol) handle() error {
switch msg.Code {
case StatusMsg:
- return ProtocolError(ErrExtraStatusMsg, "")
+ return self.protoError(ErrExtraStatusMsg, "")
case TxMsg:
// TODO: rework using lazy RLP stream
var txs []*types.Transaction
if err := msg.Decode(&txs); err != nil {
- return ProtocolError(ErrDecode, "%v", err)
+ return self.protoError(ErrDecode, "msg %v: %v", msg, err)
}
self.txPool.AddTransactions(txs)
case GetBlockHashesMsg:
var request getBlockHashesMsgData
if err := msg.Decode(&request); err != nil {
- return ProtocolError(ErrDecode, "%v", err)
+ return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
}
hashes := self.chainManager.GetBlockHashesFromHash(request.Hash, request.Amount)
return self.rw.EncodeMsg(BlockHashesMsg, ethutil.ByteSliceToInterface(hashes)...)
case BlockHashesMsg:
// TODO: redo using lazy decode , this way very inefficient on known chains
- msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size))
+ msgStream := rlp.NewStream(msg.Payload)
var err error
+ var i int
+
iter := func() (hash []byte, ok bool) {
hash, err = msgStream.Bytes()
if err == nil {
+ i++
ok = true
+ } else {
+ if err != io.EOF {
+ self.protoError(ErrDecode, "msg %v: after %v hashes : %v", msg, i, err)
+ }
}
return
}
+
self.blockPool.AddBlockHashes(iter, self.id)
- if err != nil && err != rlp.EOL {
- return ProtocolError(ErrDecode, "%v", err)
- }
case GetBlocksMsg:
- var blockHashes [][]byte
- if err := msg.Decode(&blockHashes); err != nil {
- return ProtocolError(ErrDecode, "%v", err)
- }
- max := int(math.Min(float64(len(blockHashes)), blockHashesBatchSize))
+ msgStream := rlp.NewStream(msg.Payload)
var blocks []interface{}
- for i, hash := range blockHashes {
- if i >= max {
- break
+ var i int
+ for {
+ i++
+ var hash []byte
+ if err := msgStream.Decode(&hash); err != nil {
+ if err == io.EOF {
+ break
+ } else {
+ return self.protoError(ErrDecode, "msg %v: %v", msg, err)
+ }
}
block := self.chainManager.GetBlock(hash)
if block != nil {
- blocks = append(blocks, block.RlpData())
+ blocks = append(blocks, block)
+ }
+ if i == blockHashesBatchSize {
+ break
}
}
return self.rw.EncodeMsg(BlocksMsg, blocks...)
case BlocksMsg:
- msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size))
+ msgStream := rlp.NewStream(msg.Payload)
for {
- var block *types.Block
+ var block types.Block
if err := msgStream.Decode(&block); err != nil {
- if err == rlp.EOL {
+ if err == io.EOF {
break
} else {
- return ProtocolError(ErrDecode, "%v", err)
+ return self.protoError(ErrDecode, "msg %v: %v", msg, err)
}
}
- self.blockPool.AddBlock(block, self.id)
+ self.blockPool.AddBlock(&block, self.id)
}
case NewBlockMsg:
var request newBlockMsgData
if err := msg.Decode(&request); err != nil {
- return ProtocolError(ErrDecode, "%v", err)
+ return self.protoError(ErrDecode, "msg %v: %v", msg, err)
}
hash := request.Block.Hash()
// to simplify backend interface adding a new block
@@ -202,12 +212,12 @@ func (self *ethProtocol) handle() error {
// (or selected as new best peer)
if self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) {
called := true
- iter := func() (hash []byte, ok bool) {
+ iter := func() ([]byte, bool) {
if called {
called = false
return hash, true
} else {
- return
+ return nil, false
}
}
self.blockPool.AddBlockHashes(iter, self.id)
@@ -215,14 +225,14 @@ func (self *ethProtocol) handle() error {
}
default:
- return ProtocolError(ErrInvalidMsgCode, "%v", msg.Code)
+ return self.protoError(ErrInvalidMsgCode, "%v", msg.Code)
}
return nil
}
type statusMsgData struct {
- ProtocolVersion uint
- NetworkId uint
+ ProtocolVersion uint32
+ NetworkId uint32
TD *big.Int
CurrentBlock []byte
GenesisBlock []byte
@@ -253,56 +263,56 @@ func (self *ethProtocol) handleStatus() error {
}
if msg.Code != StatusMsg {
- return ProtocolError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg)
+ return self.protoError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg)
}
if msg.Size > ProtocolMaxMsgSize {
- return ProtocolError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
+ return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
}
var status statusMsgData
if err := msg.Decode(&status); err != nil {
- return ProtocolError(ErrDecode, "%v", err)
+ return self.protoError(ErrDecode, "msg %v: %v", msg, err)
}
_, _, genesisBlock := self.chainManager.Status()
if bytes.Compare(status.GenesisBlock, genesisBlock) != 0 {
- return ProtocolError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock)
+ return self.protoError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock)
}
if status.NetworkId != NetworkId {
- return ProtocolError(ErrNetworkIdMismatch, "%d (!= %d)", status.NetworkId, NetworkId)
+ return self.protoError(ErrNetworkIdMismatch, "%d (!= %d)", status.NetworkId, NetworkId)
}
if ProtocolVersion != status.ProtocolVersion {
- return ProtocolError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, ProtocolVersion)
+ return self.protoError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, ProtocolVersion)
}
self.peer.Infof("Peer is [eth] capable (%d/%d). TD=%v H=%x\n", status.ProtocolVersion, status.NetworkId, status.TD, status.CurrentBlock[:4])
- //self.blockPool.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect)
- self.peer.Infoln("AddPeer(IGNORED)")
+ self.blockPool.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect)
return nil
}
func (self *ethProtocol) requestBlockHashes(from []byte) error {
self.peer.Debugf("fetching hashes (%d) %x...\n", blockHashesBatchSize, from[0:4])
- return self.rw.EncodeMsg(GetBlockHashesMsg, from, blockHashesBatchSize)
+ return self.rw.EncodeMsg(GetBlockHashesMsg, interface{}(from), uint64(blockHashesBatchSize))
}
func (self *ethProtocol) requestBlocks(hashes [][]byte) error {
self.peer.Debugf("fetching %v blocks", len(hashes))
- return self.rw.EncodeMsg(GetBlocksMsg, ethutil.ByteSliceToInterface(hashes))
+ return self.rw.EncodeMsg(GetBlocksMsg, ethutil.ByteSliceToInterface(hashes)...)
}
func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *protocolError) {
err = ProtocolError(code, format, params...)
if err.Fatal() {
- self.peer.Errorln(err)
+ self.peer.Errorln("err %v", err)
+ // disconnect
} else {
- self.peer.Debugln(err)
+ self.peer.Debugf("fyi %v", err)
}
return
}
@@ -310,10 +320,10 @@ func (self *ethProtocol) protoError(code int, format string, params ...interface
func (self *ethProtocol) protoErrorDisconnect(code int, format string, params ...interface{}) {
err := ProtocolError(code, format, params...)
if err.Fatal() {
- self.peer.Errorln(err)
+ self.peer.Errorln("err %v", err)
// disconnect
} else {
- self.peer.Debugln(err)
+ self.peer.Debugf("fyi %v", err)
}
}
diff --git a/eth/protocol_test.go b/eth/protocol_test.go
index 322aec7b7..ab2aa289f 100644
--- a/eth/protocol_test.go
+++ b/eth/protocol_test.go
@@ -1,35 +1,48 @@
package eth
import (
+ "bytes"
"io"
+ "log"
"math/big"
+ "os"
"testing"
+ "time"
"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"
"github.com/ethereum/go-ethereum/p2p"
)
+var sys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel))
+
type testMsgReadWriter struct {
in chan p2p.Msg
- out chan p2p.Msg
+ out []p2p.Msg
}
func (self *testMsgReadWriter) In(msg p2p.Msg) {
self.in <- msg
}
-func (self *testMsgReadWriter) Out(msg p2p.Msg) {
- self.in <- msg
+func (self *testMsgReadWriter) Out() (msg p2p.Msg, ok bool) {
+ if len(self.out) > 0 {
+ msg = self.out[0]
+ self.out = self.out[1:]
+ ok = true
+ }
+ return
}
func (self *testMsgReadWriter) WriteMsg(msg p2p.Msg) error {
- self.out <- msg
+ self.out = append(self.out, msg)
return nil
}
func (self *testMsgReadWriter) EncodeMsg(code uint64, data ...interface{}) error {
- return self.WriteMsg(p2p.NewMsg(code, data))
+ return self.WriteMsg(p2p.NewMsg(code, data...))
}
func (self *testMsgReadWriter) ReadMsg() (p2p.Msg, error) {
@@ -40,145 +53,83 @@ func (self *testMsgReadWriter) ReadMsg() (p2p.Msg, error) {
return msg, nil
}
-func errorCheck(t *testing.T, expCode int, err error) {
- perr, ok := err.(*protocolError)
- if ok && perr != nil {
- if code := perr.Code; code != expCode {
- ok = false
- }
- }
- if !ok {
- t.Errorf("expected error code %v, got %v", ErrNoStatusMsg, err)
- }
-}
-
-type TestBackend struct {
+type testTxPool struct {
getTransactions func() []*types.Transaction
addTransactions func(txs []*types.Transaction)
- getBlockHashes func(hash []byte, amount uint32) (hashes [][]byte)
- addBlockHashes func(next func() ([]byte, bool), peerId string)
- getBlock func(hash []byte) *types.Block
- addBlock func(block *types.Block, peerId string) (err error)
- addPeer func(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, invalidBlock func(error)) (best bool)
- removePeer func(peerId string)
- status func() (td *big.Int, currentBlock []byte, genesisBlock []byte)
}
-func (self *TestBackend) GetTransactions() (txs []*types.Transaction) {
- if self.getTransactions != nil {
- txs = self.getTransactions()
- }
- return
+type testChainManager struct {
+ getBlockHashes func(hash []byte, amount uint64) (hashes [][]byte)
+ getBlock func(hash []byte) *types.Block
+ status func() (td *big.Int, currentBlock []byte, genesisBlock []byte)
}
-func (self *TestBackend) AddTransactions(txs []*types.Transaction) {
+type testBlockPool struct {
+ addBlockHashes func(next func() ([]byte, bool), peerId string)
+ addBlock func(block *types.Block, peerId string) (err error)
+ addPeer func(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) (best bool)
+ removePeer func(peerId string)
+}
+
+// func (self *testTxPool) GetTransactions() (txs []*types.Transaction) {
+// if self.getTransactions != nil {
+// txs = self.getTransactions()
+// }
+// return
+// }
+
+func (self *testTxPool) AddTransactions(txs []*types.Transaction) {
if self.addTransactions != nil {
self.addTransactions(txs)
}
}
-func (self *TestBackend) GetBlockHashes(hash []byte, amount uint32) (hashes [][]byte) {
+func (self *testChainManager) GetBlockHashesFromHash(hash []byte, amount uint64) (hashes [][]byte) {
if self.getBlockHashes != nil {
hashes = self.getBlockHashes(hash, amount)
}
return
}
-<<<<<<< HEAD
-<<<<<<< HEAD
-func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId string) {
- if self.addBlockHashes != nil {
- self.addBlockHashes(next, peerId)
- }
-}
-
-=======
-func (self *TestBackend) AddHash(hash []byte, peer *p2p.Peer) (more bool) {
- if self.addHash != nil {
- more = self.addHash(hash, peer)
-=======
-func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId string) {
- if self.addBlockHashes != nil {
- self.addBlockHashes(next, peerId)
->>>>>>> eth protocol changes
+func (self *testChainManager) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) {
+ if self.status != nil {
+ td, currentBlock, genesisBlock = self.status()
}
+ return
}
-<<<<<<< HEAD
->>>>>>> initial commit for eth-p2p integration
-=======
->>>>>>> eth protocol changes
-func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) {
+func (self *testChainManager) GetBlock(hash []byte) (block *types.Block) {
if self.getBlock != nil {
block = self.getBlock(hash)
}
return
}
-<<<<<<< HEAD
-<<<<<<< HEAD
-func (self *TestBackend) AddBlock(block *types.Block, peerId string) (err error) {
- if self.addBlock != nil {
- err = self.addBlock(block, peerId)
-=======
-func (self *TestBackend) AddBlock(td *big.Int, block *types.Block, peer *p2p.Peer) (fetchHashes bool, err error) {
- if self.addBlock != nil {
- fetchHashes, err = self.addBlock(td, block, peer)
->>>>>>> initial commit for eth-p2p integration
-=======
-func (self *TestBackend) AddBlock(block *types.Block, peerId string) (err error) {
+func (self *testBlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) {
+ if self.addBlockHashes != nil {
+ self.addBlockHashes(next, peerId)
+ }
+}
+
+func (self *testBlockPool) AddBlock(block *types.Block, peerId string) {
if self.addBlock != nil {
- err = self.addBlock(block, peerId)
->>>>>>> eth protocol changes
+ self.addBlock(block, peerId)
}
- return
}
-<<<<<<< HEAD
-<<<<<<< HEAD
-func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, invalidBlock func(error)) (best bool) {
- if self.addPeer != nil {
- best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, invalidBlock)
-=======
-func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peer *p2p.Peer) (fetchHashes bool) {
+func (self *testBlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) (best bool) {
if self.addPeer != nil {
- fetchHashes = self.addPeer(td, currentBlock, peer)
->>>>>>> initial commit for eth-p2p integration
-=======
-func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, invalidBlock func(error)) (best bool) {
- if self.addPeer != nil {
- best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, invalidBlock)
->>>>>>> eth protocol changes
+ best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError)
}
return
}
-<<<<<<< HEAD
-<<<<<<< HEAD
-=======
->>>>>>> eth protocol changes
-func (self *TestBackend) RemovePeer(peerId string) {
+func (self *testBlockPool) RemovePeer(peerId string) {
if self.removePeer != nil {
self.removePeer(peerId)
}
}
-<<<<<<< HEAD
-=======
->>>>>>> initial commit for eth-p2p integration
-=======
->>>>>>> eth protocol changes
-func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) {
- if self.status != nil {
- td, currentBlock, genesisBlock = self.status()
- }
- return
-}
-
-<<<<<<< HEAD
-<<<<<<< HEAD
-=======
->>>>>>> eth protocol changes
// TODO: refactor this into p2p/client_identity
type peerId struct {
pubkey []byte
@@ -201,32 +152,119 @@ func testPeer() *p2p.Peer {
return p2p.NewPeer(&peerId{}, []p2p.Cap{})
}
-func TestErrNoStatusMsg(t *testing.T) {
-<<<<<<< HEAD
-=======
-func TestEth(t *testing.T) {
->>>>>>> initial commit for eth-p2p integration
-=======
->>>>>>> eth protocol changes
- quit := make(chan bool)
- rw := &testMsgReadWriter{make(chan p2p.Msg, 10), make(chan p2p.Msg, 10)}
- testBackend := &TestBackend{}
- var err error
- go func() {
-<<<<<<< HEAD
-<<<<<<< HEAD
- err = runEthProtocol(testBackend, testPeer(), rw)
-=======
- err = runEthProtocol(testBackend, nil, rw)
->>>>>>> initial commit for eth-p2p integration
-=======
- err = runEthProtocol(testBackend, testPeer(), rw)
->>>>>>> eth protocol changes
- close(quit)
- }()
+type ethProtocolTester struct {
+ quit chan error
+ rw *testMsgReadWriter // p2p.MsgReadWriter
+ txPool *testTxPool // txPool
+ chainManager *testChainManager // chainManager
+ blockPool *testBlockPool // blockPool
+ t *testing.T
+}
+
+func newEth(t *testing.T) *ethProtocolTester {
+ return &ethProtocolTester{
+ quit: make(chan error),
+ rw: &testMsgReadWriter{in: make(chan p2p.Msg, 10)},
+ txPool: &testTxPool{},
+ chainManager: &testChainManager{},
+ blockPool: &testBlockPool{},
+ t: t,
+ }
+}
+
+func (self *ethProtocolTester) reset() {
+ self.rw = &testMsgReadWriter{in: make(chan p2p.Msg, 10)}
+ self.quit = make(chan error)
+}
+
+func (self *ethProtocolTester) checkError(expCode int, delay time.Duration) (err error) {
+ var timer = time.After(delay)
+ select {
+ case err = <-self.quit:
+ case <-timer:
+ self.t.Errorf("no error after %v, expected %v", delay, expCode)
+ return
+ }
+ perr, ok := err.(*protocolError)
+ if ok && perr != nil {
+ if code := perr.Code; code != expCode {
+ self.t.Errorf("expected protocol error (code %v), got %v (%v)", expCode, code, err)
+ }
+ } else {
+ self.t.Errorf("expected protocol error (code %v), got %v", expCode, err)
+ }
+ return
+}
+
+func (self *ethProtocolTester) In(msg p2p.Msg) {
+ self.rw.In(msg)
+}
+
+func (self *ethProtocolTester) Out() (p2p.Msg, bool) {
+ return self.rw.Out()
+}
+
+func (self *ethProtocolTester) checkMsg(i int, code uint64, val interface{}) (msg p2p.Msg) {
+ if i >= len(self.rw.out) {
+ self.t.Errorf("expected at least %v msgs, got %v", i, len(self.rw.out))
+ return
+ }
+ msg = self.rw.out[i]
+ if msg.Code != code {
+ self.t.Errorf("expected msg code %v, got %v", code, msg.Code)
+ }
+ if val != nil {
+ if err := msg.Decode(val); err != nil {
+ self.t.Errorf("rlp encoding error: %v", err)
+ }
+ }
+ return
+}
+
+func (self *ethProtocolTester) run() {
+ err := runEthProtocol(self.txPool, self.chainManager, self.blockPool, testPeer(), self.rw)
+ self.quit <- err
+}
+
+func TestStatusMsgErrors(t *testing.T) {
+ logInit()
+ eth := newEth(t)
+ td := ethutil.Big1
+ currentBlock := []byte{1}
+ genesis := []byte{2}
+ eth.chainManager.status = func() (*big.Int, []byte, []byte) { return td, currentBlock, genesis }
+ go eth.run()
statusMsg := p2p.NewMsg(4)
- rw.In(statusMsg)
- <-quit
- errorCheck(t, ErrNoStatusMsg, err)
- // read(t, remote, []byte("hello, world"), nil)
+ eth.In(statusMsg)
+ delay := 1 * time.Second
+ eth.checkError(ErrNoStatusMsg, delay)
+ var status statusMsgData
+ eth.checkMsg(0, StatusMsg, &status) // first outgoing msg should be StatusMsg
+ if status.TD.Cmp(td) != 0 ||
+ status.ProtocolVersion != ProtocolVersion ||
+ status.NetworkId != NetworkId ||
+ status.TD.Cmp(td) != 0 ||
+ bytes.Compare(status.CurrentBlock, currentBlock) != 0 ||
+ bytes.Compare(status.GenesisBlock, genesis) != 0 {
+ t.Errorf("incorrect outgoing status")
+ }
+
+ eth.reset()
+ go eth.run()
+ statusMsg = p2p.NewMsg(0, uint32(48), uint32(0), td, currentBlock, genesis)
+ eth.In(statusMsg)
+ eth.checkError(ErrProtocolVersionMismatch, delay)
+
+ eth.reset()
+ go eth.run()
+ statusMsg = p2p.NewMsg(0, uint32(49), uint32(1), td, currentBlock, genesis)
+ eth.In(statusMsg)
+ eth.checkError(ErrNetworkIdMismatch, delay)
+
+ eth.reset()
+ go eth.run()
+ statusMsg = p2p.NewMsg(0, uint32(49), uint32(0), td, currentBlock, []byte{3})
+ eth.In(statusMsg)
+ eth.checkError(ErrGenesisBlockMismatch, delay)
+
}
diff --git a/eth/test/README.md b/eth/test/README.md
new file mode 100644
index 000000000..65728efa5
--- /dev/null
+++ b/eth/test/README.md
@@ -0,0 +1,27 @@
+= Integration tests for eth protocol and blockpool
+
+This is a simple suite of tests to fire up a local test node with peers to test blockchain synchronisation and download.
+The scripts call ethereum (assumed to be compiled in go-ethereum root).
+
+To run a test:
+
+ . run.sh 00 02
+
+Without arguments, all tests are run.
+
+Peers are launched with preloaded imported chains. In order to prevent them from synchronizing with each other they are set with `-dial=false` and `-maxpeer 1` options. They log into `/tmp/eth.test/nodes/XX` where XX is the last two digits of their port.
+
+Chains to import can be bootstrapped by letting nodes mine for some time. This is done with
+
+ . bootstrap.sh
+
+Only the relative timing and forks matter so they should work if the bootstrap script is rerun.
+The reference blockchain of tests are soft links to these import chains and check at the end of a test run.
+
+Connecting to peers and exporting blockchain is scripted with JS files executed by the JSRE, see `tests/XX.sh`.
+
+Each test is set with a timeout. This may vary on different computers so adjust sensibly.
+If you kill a test before it completes, do not forget to kill all the background processes, since they will impact the result. Use:
+
+ killall ethereum
+
diff --git a/eth/test/bootstrap.sh b/eth/test/bootstrap.sh
new file mode 100644
index 000000000..3da038be8
--- /dev/null
+++ b/eth/test/bootstrap.sh
@@ -0,0 +1,9 @@
+#!/bin/bash
+# bootstrap chains - used to regenerate tests/chains/*.chain
+
+mkdir -p chains
+bash ./mine.sh 00 10
+bash ./mine.sh 01 5 00
+bash ./mine.sh 02 10 00
+bash ./mine.sh 03 5 02
+bash ./mine.sh 04 10 02 \ No newline at end of file
diff --git a/eth/test/chains/00.chain b/eth/test/chains/00.chain
new file mode 100755
index 000000000..ad3c05b24
--- /dev/null
+++ b/eth/test/chains/00.chain
Binary files differ
diff --git a/eth/test/chains/01.chain b/eth/test/chains/01.chain
new file mode 100755
index 000000000..56c9aef65
--- /dev/null
+++ b/eth/test/chains/01.chain
Binary files differ
diff --git a/eth/test/chains/02.chain b/eth/test/chains/02.chain
new file mode 100755
index 000000000..440c92d65
--- /dev/null
+++ b/eth/test/chains/02.chain
Binary files differ
diff --git a/eth/test/chains/03.chain b/eth/test/chains/03.chain
new file mode 100755
index 000000000..1cc7570ab
--- /dev/null
+++ b/eth/test/chains/03.chain
Binary files differ
diff --git a/eth/test/chains/04.chain b/eth/test/chains/04.chain
new file mode 100755
index 000000000..d4e5b1aa8
--- /dev/null
+++ b/eth/test/chains/04.chain
Binary files differ
diff --git a/eth/test/mine.sh b/eth/test/mine.sh
new file mode 100644
index 000000000..0d95db1e4
--- /dev/null
+++ b/eth/test/mine.sh
@@ -0,0 +1,20 @@
+#!/bin/bash
+# bash ./mine.sh node_id timeout(sec) [basechain]
+ETH=../../ethereum
+MINE="$ETH -datadir tmp/nodes/$1 -seed=false -port '' -shh=false -id test$1"
+rm -rf tmp/nodes/$1
+echo "Creating chain $1..."
+if [[ "" != "$3" ]]; then
+ CHAIN="chains/$3.chain"
+ CHAINARG="-chain $CHAIN"
+ $MINE -mine $CHAINARG -loglevel 3 | grep 'importing'
+fi
+$MINE -mine -loglevel 0 &
+PID=$!
+sleep $2
+kill $PID
+$MINE -loglevel 3 <(echo "eth.export(\"chains/$1.chain\")") > /tmp/eth.test/mine.tmp &
+PID=$!
+sleep 1
+kill $PID
+cat /tmp/eth.test/mine.tmp | grep 'exporting'
diff --git a/eth/test/run.sh b/eth/test/run.sh
new file mode 100644
index 000000000..5229af035
--- /dev/null
+++ b/eth/test/run.sh
@@ -0,0 +1,53 @@
+#!/bin/bash
+# bash run.sh (testid0 testid1 ...)
+# runs tests tests/testid0.sh tests/testid1.sh ...
+# without arguments, it runs all tests
+
+. tests/common.sh
+
+TESTS=
+
+if [ "$#" -eq 0 ]; then
+ for NAME in tests/??.sh; do
+ i=`basename $NAME .sh`
+ TESTS="$TESTS $i"
+ done
+else
+ TESTS=$@
+fi
+
+ETH=../../ethereum
+DIR="/tmp/eth.test/nodes"
+TIMEOUT=10
+
+mkdir -p $DIR/js
+
+echo "running tests $TESTS"
+for NAME in $TESTS; do
+ PIDS=
+ CHAIN="tests/$NAME.chain"
+ JSFILE="$DIR/js/$NAME.js"
+ CHAIN_TEST="$DIR/$NAME/chain"
+
+ echo "RUN: test $NAME"
+ cat tests/common.js > $JSFILE
+ . tests/$NAME.sh
+ sleep $TIMEOUT
+ echo "timeout after $TIMEOUT seconds: killing $PIDS"
+ kill $PIDS
+ if [ -r "$CHAIN" ]; then
+ if diff $CHAIN $CHAIN_TEST >/dev/null ; then
+ echo "chain ok: $CHAIN=$CHAIN_TEST"
+ else
+ echo "FAIL: chains differ: expected $CHAIN ; got $CHAIN_TEST"
+ continue
+ fi
+ fi
+ ERRORS=$DIR/errors
+ if [ -r "$ERRORS" ]; then
+ echo "FAIL: "
+ cat $ERRORS
+ else
+ echo PASS
+ fi
+done \ No newline at end of file
diff --git a/eth/test/tests/00.chain b/eth/test/tests/00.chain
new file mode 120000
index 000000000..9655cb3df
--- /dev/null
+++ b/eth/test/tests/00.chain
@@ -0,0 +1 @@
+../chains/01.chain \ No newline at end of file
diff --git a/eth/test/tests/00.sh b/eth/test/tests/00.sh
new file mode 100644
index 000000000..9c5077164
--- /dev/null
+++ b/eth/test/tests/00.sh
@@ -0,0 +1,13 @@
+#!/bin/bash
+
+TIMEOUT=4
+
+cat >> $JSFILE <<EOF
+eth.addPeer("localhost:30311");
+sleep(1000)
+eth.export("$CHAIN_TEST");
+EOF
+
+peer 11 01
+test_node $NAME "" -loglevel 5 $JSFILE
+
diff --git a/eth/test/tests/01.chain b/eth/test/tests/01.chain
new file mode 120000
index 000000000..ae65ccb37
--- /dev/null
+++ b/eth/test/tests/01.chain
@@ -0,0 +1 @@
+../chains/02.chain \ No newline at end of file
diff --git a/eth/test/tests/01.sh b/eth/test/tests/01.sh
new file mode 100644
index 000000000..1601e0dfb
--- /dev/null
+++ b/eth/test/tests/01.sh
@@ -0,0 +1,18 @@
+#!/bin/bash
+
+TIMEOUT=5
+
+cat >> $JSFILE <<EOF
+eth.addPeer("localhost:30311");
+log("added peer localhost:30311");
+sleep(1000);
+log("added peer localhost:30312");
+eth.addPeer("localhost:30312");
+sleep(3000);
+eth.export("$CHAIN_TEST");
+EOF
+
+peer 11 01
+peer 12 02
+test_node $NAME "" -loglevel 5 $JSFILE
+
diff --git a/eth/test/tests/02.chain b/eth/test/tests/02.chain
new file mode 120000
index 000000000..9655cb3df
--- /dev/null
+++ b/eth/test/tests/02.chain
@@ -0,0 +1 @@
+../chains/01.chain \ No newline at end of file
diff --git a/eth/test/tests/02.sh b/eth/test/tests/02.sh
new file mode 100644
index 000000000..5231dbd78
--- /dev/null
+++ b/eth/test/tests/02.sh
@@ -0,0 +1,19 @@
+#!/bin/bash
+
+TIMEOUT=6
+
+cat >> $JSFILE <<EOF
+eth.addPeer("localhost:30311");
+sleep(200);
+eth.addPeer("localhost:30312");
+sleep(3000);
+eth.export("$CHAIN_TEST");
+EOF
+
+peer 11 01
+peer 12 02
+P13ID=$PID
+test_node $NAME "" -loglevel 5 $JSFILE
+sleep 0.5
+kill $P13ID
+
diff --git a/eth/test/tests/03.chain b/eth/test/tests/03.chain
new file mode 120000
index 000000000..b07c49a30
--- /dev/null
+++ b/eth/test/tests/03.chain
@@ -0,0 +1 @@
+../chains/12k.chain \ No newline at end of file
diff --git a/eth/test/tests/03.sh b/eth/test/tests/03.sh
new file mode 100644
index 000000000..8c9d6565e
--- /dev/null
+++ b/eth/test/tests/03.sh
@@ -0,0 +1,14 @@
+#!/bin/bash
+
+TIMEOUT=35
+
+cat >> $JSFILE <<EOF
+eth.addPeer("localhost:30311");
+sleep(30000);
+eth.export("$CHAIN_TEST");
+EOF
+
+peer 11 12k
+sleep 2
+test_node $NAME "" -loglevel 5 $JSFILE
+
diff --git a/eth/test/tests/04.sh b/eth/test/tests/04.sh
new file mode 100644
index 000000000..d77c360ba
--- /dev/null
+++ b/eth/test/tests/04.sh
@@ -0,0 +1,17 @@
+#!/bin/bash
+
+TIMEOUT=15
+
+cat >> $JSFILE <<EOF
+eth.addPeer("localhost:30311");
+sleep(200);
+eth.addPeer("localhost:30312");
+sleep(13000);
+eth.export("$CHAIN_TEST");
+EOF
+
+peer 11 01 -mine
+peer 12 02
+test_node $NAME "" -loglevel 5 $JSFILE
+sleep 6
+cat $DIR/$NAME/debug.log | grep 'best peer'
diff --git a/eth/test/tests/05.sh b/eth/test/tests/05.sh
new file mode 100644
index 000000000..0a0f94df5
--- /dev/null
+++ b/eth/test/tests/05.sh
@@ -0,0 +1,20 @@
+#!/bin/bash
+
+TIMEOUT=60
+
+cat >> $JSFILE <<EOF
+eth.addPeer("localhost:30311");
+sleep(200);
+eth.addPeer("localhost:30312");
+eth.addPeer("localhost:30313");
+eth.addPeer("localhost:30314");
+sleep(3000);
+eth.export("$CHAIN_TEST");
+EOF
+
+peer 11 01 -mine
+peer 12 02 -mine
+peer 13 03
+peer 14 04
+test_node $NAME "" -loglevel 5 $JSFILE
+
diff --git a/eth/test/tests/common.js b/eth/test/tests/common.js
new file mode 100644
index 000000000..206ebf145
--- /dev/null
+++ b/eth/test/tests/common.js
@@ -0,0 +1,9 @@
+function log(text) {
+ console.log("[JS TEST SCRIPT] " + text);
+}
+
+function sleep(seconds) {
+ var now = new Date().getTime();
+ while(new Date().getTime() < now + seconds){}
+}
+
diff --git a/eth/test/tests/common.sh b/eth/test/tests/common.sh
new file mode 100644
index 000000000..74db073f7
--- /dev/null
+++ b/eth/test/tests/common.sh
@@ -0,0 +1,20 @@
+#!/bin/bash
+
+# launched by run.sh
+function test_node {
+ rm -rf $DIR/$1
+ ARGS="-datadir $DIR/$1 -debug debug -seed=false -shh=false -id test$1"
+ if [ "" != "$2" ]; then
+ chain="chains/$2.chain"
+ echo "import chain $chain"
+ $ETH $ARGS -loglevel 3 -chain $chain | grep CLI |grep import
+ fi
+ echo "starting test node $1 with extra args ${@:3}"
+ $ETH $ARGS -port 303$1 ${@:3} &
+ PID=$!
+ PIDS="$PIDS $PID"
+}
+
+function peer {
+ test_node $@ -loglevel 5 -logfile debug.log -maxpeer 1 -dial=false
+} \ No newline at end of file
diff --git a/p2p/client_identity_test.go b/p2p/client_identity_test.go
index 40b0e6f5e..7248a7b1a 100644
--- a/p2p/client_identity_test.go
+++ b/p2p/client_identity_test.go
@@ -7,7 +7,7 @@ import (
)
func TestClientIdentity(t *testing.T) {
- clientIdentity := NewSimpleClientIdentity("Ethereum(G)", "0.5.16", "test", "pubkey")
+ clientIdentity := NewSimpleClientIdentity("Ethereum(G)", "0.5.16", "test", []byte("pubkey"))
clientString := clientIdentity.String()
expected := fmt.Sprintf("Ethereum(G)/v0.5.16/test/%s/%s", runtime.GOOS, runtime.Version())
if clientString != expected {
diff --git a/p2p/message.go b/p2p/message.go
index f5418ff47..a6f62ec4c 100644
--- a/p2p/message.go
+++ b/p2p/message.go
@@ -4,6 +4,7 @@ import (
"bytes"
"encoding/binary"
"errors"
+ "fmt"
"io"
"io/ioutil"
"math/big"
@@ -49,7 +50,14 @@ func encodePayload(params ...interface{}) []byte {
// For the decoding rules, please see package rlp.
func (msg Msg) Decode(val interface{}) error {
s := rlp.NewListStream(msg.Payload, uint64(msg.Size))
- return s.Decode(val)
+ if err := s.Decode(val); err != nil {
+ return newPeerError(errInvalidMsg, "(code %#x) (size %d) %v", msg.Code, msg.Size, err)
+ }
+ return nil
+}
+
+func (msg Msg) String() string {
+ return fmt.Sprintf("msg #%v (%v bytes)", msg.Code, msg.Size)
}
// Discard reads any remaining payload data into a black hole.
diff --git a/p2p/peer.go b/p2p/peer.go
index 86c4d7ab5..0d7eec9f4 100644
--- a/p2p/peer.go
+++ b/p2p/peer.go
@@ -45,8 +45,8 @@ func (d peerAddr) String() string {
return fmt.Sprintf("%v:%d", d.IP, d.Port)
}
-func (d peerAddr) RlpData() interface{} {
- return []interface{}{d.IP, d.Port, d.Pubkey}
+func (d *peerAddr) RlpData() interface{} {
+ return []interface{}{string(d.IP), d.Port, d.Pubkey}
}
// Peer represents a remote peer.
@@ -426,7 +426,7 @@ func (rw *proto) WriteMsg(msg Msg) error {
}
func (rw *proto) EncodeMsg(code uint64, data ...interface{}) error {
- return rw.WriteMsg(NewMsg(code, data))
+ return rw.WriteMsg(NewMsg(code, data...))
}
func (rw *proto) ReadMsg() (Msg, error) {
@@ -460,3 +460,25 @@ func (r *eofSignal) Read(buf []byte) (int, error) {
}
return n, err
}
+
+func (peer *Peer) PeerList() []interface{} {
+ peers := peer.otherPeers()
+ ds := make([]interface{}, 0, len(peers))
+ for _, p := range peers {
+ p.infolock.Lock()
+ addr := p.listenAddr
+ p.infolock.Unlock()
+ // filter out this peer and peers that are not listening or
+ // have not completed the handshake.
+ // TODO: track previously sent peers and exclude them as well.
+ if p == peer || addr == nil {
+ continue
+ }
+ ds = append(ds, addr)
+ }
+ ourAddr := peer.ourListenAddr
+ if ourAddr != nil && !ourAddr.IP.IsLoopback() && !ourAddr.IP.IsUnspecified() {
+ ds = append(ds, ourAddr)
+ }
+ return ds
+}
diff --git a/p2p/peer_test.go b/p2p/peer_test.go
index f7759786e..5b9e9e784 100644
--- a/p2p/peer_test.go
+++ b/p2p/peer_test.go
@@ -30,9 +30,8 @@ var discard = Protocol{
func testPeer(protos []Protocol) (net.Conn, *Peer, <-chan error) {
conn1, conn2 := net.Pipe()
- id := NewSimpleClientIdentity("test", "0", "0", "public key")
peer := newPeer(conn1, protos, nil)
- peer.ourID = id
+ peer.ourID = &peerId{}
peer.pubkeyHook = func(*peerAddr) error { return nil }
errc := make(chan error, 1)
go func() {
@@ -130,7 +129,7 @@ func TestPeerProtoEncodeMsg(t *testing.T) {
if err := rw.EncodeMsg(2); err == nil {
t.Error("expected error for out-of-range msg code, got nil")
}
- if err := rw.EncodeMsg(1); err != nil {
+ if err := rw.EncodeMsg(1, "foo", "bar"); err != nil {
t.Errorf("write error: %v", err)
}
return nil
@@ -148,6 +147,13 @@ func TestPeerProtoEncodeMsg(t *testing.T) {
if msg.Code != 17 {
t.Errorf("incorrect message code: got %d, expected %d", msg.Code, 17)
}
+ var data []string
+ if err := msg.Decode(&data); err != nil {
+ t.Errorf("payload decode error: %v", err)
+ }
+ if !reflect.DeepEqual(data, []string{"foo", "bar"}) {
+ t.Errorf("payload RLP mismatch, got %#v, want %#v", data, []string{"foo", "bar"})
+ }
}
func TestPeerWrite(t *testing.T) {
@@ -226,8 +232,8 @@ func TestPeerActivity(t *testing.T) {
}
func TestNewPeer(t *testing.T) {
- id := NewSimpleClientIdentity("clientid", "version", "customid", "pubkey")
caps := []Cap{{"foo", 2}, {"bar", 3}}
+ id := &peerId{}
p := NewPeer(id, caps)
if !reflect.DeepEqual(p.Caps(), caps) {
t.Errorf("Caps mismatch: got %v, expected %v", p.Caps(), caps)
diff --git a/p2p/protocol.go b/p2p/protocol.go
index 3f52205f5..dd8cbc4ec 100644
--- a/p2p/protocol.go
+++ b/p2p/protocol.go
@@ -3,8 +3,6 @@ package p2p
import (
"bytes"
"time"
-
- "github.com/ethereum/go-ethereum/ethutil"
)
// Protocol represents a P2P subprotocol implementation.
@@ -89,20 +87,25 @@ type baseProtocol struct {
func runBaseProtocol(peer *Peer, rw MsgReadWriter) error {
bp := &baseProtocol{rw, peer}
- if err := bp.doHandshake(rw); err != nil {
+ errc := make(chan error, 1)
+ go func() { errc <- rw.WriteMsg(bp.handshakeMsg()) }()
+ if err := bp.readHandshake(); err != nil {
+ return err
+ }
+ // handle write error
+ if err := <-errc; err != nil {
return err
}
// run main loop
- quit := make(chan error, 1)
go func() {
for {
if err := bp.handle(rw); err != nil {
- quit <- err
+ errc <- err
break
}
}
}()
- return bp.loop(quit)
+ return bp.loop(errc)
}
var pingTimeout = 2 * time.Second
@@ -166,7 +169,7 @@ func (bp *baseProtocol) handle(rw MsgReadWriter) error {
case pongMsg:
case getPeersMsg:
- peers := bp.peerList()
+ peers := bp.peer.PeerList()
// this is dangerous. the spec says that we should _delay_
// sending the response if no new information is available.
// this means that would need to send a response later when
@@ -174,7 +177,7 @@ func (bp *baseProtocol) handle(rw MsgReadWriter) error {
//
// TODO: add event mechanism to notify baseProtocol for new peers
if len(peers) > 0 {
- return bp.rw.EncodeMsg(peersMsg, peers)
+ return bp.rw.EncodeMsg(peersMsg, peers...)
}
case peersMsg:
@@ -193,14 +196,9 @@ func (bp *baseProtocol) handle(rw MsgReadWriter) error {
return nil
}
-func (bp *baseProtocol) doHandshake(rw MsgReadWriter) error {
- // send our handshake
- if err := rw.WriteMsg(bp.handshakeMsg()); err != nil {
- return err
- }
-
+func (bp *baseProtocol) readHandshake() error {
// read and handle remote handshake
- msg, err := rw.ReadMsg()
+ msg, err := bp.rw.ReadMsg()
if err != nil {
return err
}
@@ -210,12 +208,10 @@ func (bp *baseProtocol) doHandshake(rw MsgReadWriter) error {
if msg.Size > baseProtocolMaxMsgSize {
return newPeerError(errMisc, "message too big")
}
-
var hs handshake
if err := msg.Decode(&hs); err != nil {
return err
}
-
// validate handshake info
if hs.Version != baseProtocolVersion {
return newPeerError(errP2PVersionMismatch, "Require protocol %d, received %d\n",
@@ -238,9 +234,7 @@ func (bp *baseProtocol) doHandshake(rw MsgReadWriter) error {
if err := bp.peer.pubkeyHook(pa); err != nil {
return newPeerError(errPubkeyForbidden, "%v", err)
}
-
// TODO: remove Caps with empty name
-
var addr *peerAddr
if hs.ListenPort != 0 {
addr = newPeerAddr(bp.peer.conn.RemoteAddr(), hs.NodeID)
@@ -270,25 +264,3 @@ func (bp *baseProtocol) handshakeMsg() Msg {
bp.peer.ourID.Pubkey()[1:],
)
}
-
-func (bp *baseProtocol) peerList() []ethutil.RlpEncodable {
- peers := bp.peer.otherPeers()
- ds := make([]ethutil.RlpEncodable, 0, len(peers))
- for _, p := range peers {
- p.infolock.Lock()
- addr := p.listenAddr
- p.infolock.Unlock()
- // filter out this peer and peers that are not listening or
- // have not completed the handshake.
- // TODO: track previously sent peers and exclude them as well.
- if p == bp.peer || addr == nil {
- continue
- }
- ds = append(ds, addr)
- }
- ourAddr := bp.peer.ourListenAddr
- if ourAddr != nil && !ourAddr.IP.IsLoopback() && !ourAddr.IP.IsUnspecified() {
- ds = append(ds, ourAddr)
- }
- return ds
-}
diff --git a/p2p/protocol_test.go b/p2p/protocol_test.go
index 65f26fb12..ce25b3e1b 100644
--- a/p2p/protocol_test.go
+++ b/p2p/protocol_test.go
@@ -2,12 +2,89 @@ package p2p
import (
"fmt"
+ "net"
+ "reflect"
"testing"
+
+ "github.com/ethereum/go-ethereum/crypto"
)
+type peerId struct {
+ pubkey []byte
+}
+
+func (self *peerId) String() string {
+ return fmt.Sprintf("test peer %x", self.Pubkey()[:4])
+}
+
+func (self *peerId) Pubkey() (pubkey []byte) {
+ pubkey = self.pubkey
+ if len(pubkey) == 0 {
+ pubkey = crypto.GenerateNewKeyPair().PublicKey
+ self.pubkey = pubkey
+ }
+ return
+}
+
+func newTestPeer() (peer *Peer) {
+ peer = NewPeer(&peerId{}, []Cap{})
+ peer.pubkeyHook = func(*peerAddr) error { return nil }
+ peer.ourID = &peerId{}
+ peer.listenAddr = &peerAddr{}
+ peer.otherPeers = func() []*Peer { return nil }
+ return
+}
+
+func TestBaseProtocolPeers(t *testing.T) {
+ cannedPeerList := []*peerAddr{
+ {IP: net.ParseIP("1.2.3.4"), Port: 2222, Pubkey: []byte{}},
+ {IP: net.ParseIP("5.6.7.8"), Port: 3333, Pubkey: []byte{}},
+ }
+ var ownAddr *peerAddr = &peerAddr{IP: net.ParseIP("1.3.5.7"), Port: 1111, Pubkey: []byte{}}
+ rw1, rw2 := MsgPipe()
+ // run matcher, close pipe when addresses have arrived
+ addrChan := make(chan *peerAddr, len(cannedPeerList))
+ go func() {
+ for _, want := range cannedPeerList {
+ got := <-addrChan
+ t.Logf("got peer: %+v", got)
+ if !reflect.DeepEqual(want, got) {
+ t.Errorf("mismatch: got %#v, want %#v", got, want)
+ }
+ }
+ close(addrChan)
+ var own []*peerAddr
+ var got *peerAddr
+ for got = range addrChan {
+ own = append(own, got)
+ }
+ if len(own) != 1 || !reflect.DeepEqual(ownAddr, own[0]) {
+ t.Errorf("mismatch: peers own address is incorrectly or not given, got %v, want %#v", ownAddr)
+ }
+ rw2.Close()
+ }()
+ // run first peer
+ peer1 := newTestPeer()
+ peer1.ourListenAddr = ownAddr
+ peer1.otherPeers = func() []*Peer {
+ pl := make([]*Peer, len(cannedPeerList))
+ for i, addr := range cannedPeerList {
+ pl[i] = &Peer{listenAddr: addr}
+ }
+ return pl
+ }
+ go runBaseProtocol(peer1, rw1)
+ // run second peer
+ peer2 := newTestPeer()
+ peer2.newPeerAddr = addrChan // feed peer suggestions into matcher
+ if err := runBaseProtocol(peer2, rw2); err != ErrPipeClosed {
+ t.Errorf("peer2 terminated with unexpected error: %v", err)
+ }
+}
+
func TestBaseProtocolDisconnect(t *testing.T) {
- peer := NewPeer(NewSimpleClientIdentity("p1", "", "", "foo"), nil)
- peer.ourID = NewSimpleClientIdentity("p2", "", "", "bar")
+ peer := NewPeer(&peerId{}, nil)
+ peer.ourID = &peerId{}
peer.pubkeyHook = func(*peerAddr) error { return nil }
rw1, rw2 := MsgPipe()
@@ -32,6 +109,7 @@ func TestBaseProtocolDisconnect(t *testing.T) {
if err := rw2.EncodeMsg(discMsg, DiscQuitting); err != nil {
t.Error(err)
}
+
close(done)
}()
diff --git a/p2p/server.go b/p2p/server.go
index 326781234..cfff442f7 100644
--- a/p2p/server.go
+++ b/p2p/server.go
@@ -113,9 +113,11 @@ func (srv *Server) PeerCount() int {
// SuggestPeer injects an address into the outbound address pool.
func (srv *Server) SuggestPeer(ip net.IP, port int, nodeID []byte) {
+ addr := &peerAddr{ip, uint64(port), nodeID}
select {
- case srv.peerConnect <- &peerAddr{ip, uint64(port), nodeID}:
+ case srv.peerConnect <- addr:
default: // don't block
+ srvlog.Warnf("peer suggestion %v ignored", addr)
}
}
@@ -258,6 +260,7 @@ func (srv *Server) listenLoop() {
for {
select {
case slot := <-srv.peerSlots:
+ srvlog.Debugf("grabbed slot %v for listening", slot)
conn, err := srv.listener.Accept()
if err != nil {
srv.peerSlots <- slot
@@ -330,6 +333,7 @@ func (srv *Server) dialLoop() {
case desc := <-suggest:
// candidate peer found, will dial out asyncronously
// if connection fails slot will be released
+ srvlog.Infof("dial %v (%v)", desc, *slot)
go srv.dialPeer(desc, *slot)
// we can watch if more peers needed in the next loop
slots = srv.peerSlots
diff --git a/p2p/server_test.go b/p2p/server_test.go
index 5c0d08d39..ceb89e3f7 100644
--- a/p2p/server_test.go
+++ b/p2p/server_test.go
@@ -11,7 +11,7 @@ import (
func startTestServer(t *testing.T, pf peerFunc) *Server {
server := &Server{
- Identity: NewSimpleClientIdentity("clientIdentifier", "version", "customIdentifier", "pubkey"),
+ Identity: &peerId{},
MaxPeers: 10,
ListenAddr: "127.0.0.1:0",
newPeerFunc: pf,
diff --git a/rlp/decode.go b/rlp/decode.go
index 712d9fcf1..a2bd04285 100644
--- a/rlp/decode.go
+++ b/rlp/decode.go
@@ -76,22 +76,37 @@ func Decode(r io.Reader, val interface{}) error {
type decodeError struct {
msg string
typ reflect.Type
+ ctx []string
}
-func (err decodeError) Error() string {
- return fmt.Sprintf("rlp: %s for %v", err.msg, err.typ)
+func (err *decodeError) Error() string {
+ ctx := ""
+ if len(err.ctx) > 0 {
+ ctx = ", decoding into "
+ for i := len(err.ctx) - 1; i >= 0; i-- {
+ ctx += err.ctx[i]
+ }
+ }
+ return fmt.Sprintf("rlp: %s for %v%s", err.msg, err.typ, ctx)
}
func wrapStreamError(err error, typ reflect.Type) error {
switch err {
case ErrExpectedList:
- return decodeError{"expected input list", typ}
+ return &decodeError{msg: "expected input list", typ: typ}
case ErrExpectedString:
- return decodeError{"expected input string or byte", typ}
+ return &decodeError{msg: "expected input string or byte", typ: typ}
case errUintOverflow:
- return decodeError{"input string too long", typ}
+ return &decodeError{msg: "input string too long", typ: typ}
case errNotAtEOL:
- return decodeError{"input list has too many elements", typ}
+ return &decodeError{msg: "input list has too many elements", typ: typ}
+ }
+ return err
+}
+
+func addErrorContext(err error, ctx string) error {
+ if decErr, ok := err.(*decodeError); ok {
+ decErr.ctx = append(decErr.ctx, ctx)
}
return err
}
@@ -180,13 +195,13 @@ func makeListDecoder(typ reflect.Type) (decoder, error) {
return nil, err
}
- if typ.Kind() == reflect.Array {
- return func(s *Stream, val reflect.Value) error {
- return decodeListArray(s, val, etypeinfo.decoder)
- }, nil
- }
+ isArray := typ.Kind() == reflect.Array
return func(s *Stream, val reflect.Value) error {
- return decodeListSlice(s, val, etypeinfo.decoder)
+ if isArray {
+ return decodeListArray(s, val, etypeinfo.decoder)
+ } else {
+ return decodeListSlice(s, val, etypeinfo.decoder)
+ }
}, nil
}
@@ -219,7 +234,7 @@ func decodeListSlice(s *Stream, val reflect.Value, elemdec decoder) error {
if err := elemdec(s, val.Index(i)); err == EOL {
break
} else if err != nil {
- return err
+ return addErrorContext(err, fmt.Sprint("[", i, "]"))
}
}
if i < val.Len() {
@@ -248,7 +263,7 @@ func decodeListArray(s *Stream, val reflect.Value, elemdec decoder) error {
if err := elemdec(s, val.Index(i)); err == EOL {
break
} else if err != nil {
- return err
+ return addErrorContext(err, fmt.Sprint("[", i, "]"))
}
}
if i < vlen {
@@ -280,14 +295,14 @@ func decodeByteArray(s *Stream, val reflect.Value) error {
switch kind {
case Byte:
if val.Len() == 0 {
- return decodeError{"input string too long", val.Type()}
+ return &decodeError{msg: "input string too long", typ: val.Type()}
}
bv, _ := s.Uint()
val.Index(0).SetUint(bv)
zero(val, 1)
case String:
if uint64(val.Len()) < size {
- return decodeError{"input string too long", val.Type()}
+ return &decodeError{msg: "input string too long", typ: val.Type()}
}
slice := val.Slice(0, int(size)).Interface().([]byte)
if err := s.readFull(slice); err != nil {
@@ -334,7 +349,7 @@ func makeStructDecoder(typ reflect.Type) (decoder, error) {
// too few elements. leave the rest at their zero value.
break
} else if err != nil {
- return err
+ return addErrorContext(err, "."+typ.Field(f.index).Name)
}
}
return wrapStreamError(s.ListEnd(), typ)
@@ -599,7 +614,13 @@ func (s *Stream) Decode(val interface{}) error {
if err != nil {
return err
}
- return info.decoder(s, rval.Elem())
+
+ err = info.decoder(s, rval.Elem())
+ if decErr, ok := err.(*decodeError); ok && len(decErr.ctx) > 0 {
+ // add decode target type to error so context has more meaning
+ decErr.ctx = append(decErr.ctx, fmt.Sprint("(", rtyp.Elem(), ")"))
+ }
+ return err
}
// Reset discards any information about the current decoding context
diff --git a/rlp/decode_test.go b/rlp/decode_test.go
index 7a1743937..18ea63a09 100644
--- a/rlp/decode_test.go
+++ b/rlp/decode_test.go
@@ -231,7 +231,12 @@ var decodeTests = []decodeTest{
{input: "8D6162636465666768696A6B6C6D", ptr: new([]byte), value: []byte("abcdefghijklm")},
{input: "C0", ptr: new([]byte), value: []byte{}},
{input: "C3010203", ptr: new([]byte), value: []byte{1, 2, 3}},
- {input: "C3820102", ptr: new([]byte), error: "rlp: input string too long for uint8"},
+
+ {
+ input: "C3820102",
+ ptr: new([]byte),
+ error: "rlp: input string too long for uint8, decoding into ([]uint8)[0]",
+ },
// byte arrays
{input: "01", ptr: new([5]byte), value: [5]byte{1}},
@@ -239,9 +244,22 @@ var decodeTests = []decodeTest{
{input: "850102030405", ptr: new([5]byte), value: [5]byte{1, 2, 3, 4, 5}},
{input: "C0", ptr: new([5]byte), value: [5]byte{}},
{input: "C3010203", ptr: new([5]byte), value: [5]byte{1, 2, 3, 0, 0}},
- {input: "C3820102", ptr: new([5]byte), error: "rlp: input string too long for uint8"},
- {input: "86010203040506", ptr: new([5]byte), error: "rlp: input string too long for [5]uint8"},
- {input: "850101", ptr: new([5]byte), error: io.ErrUnexpectedEOF.Error()},
+
+ {
+ input: "C3820102",
+ ptr: new([5]byte),
+ error: "rlp: input string too long for uint8, decoding into ([5]uint8)[0]",
+ },
+ {
+ input: "86010203040506",
+ ptr: new([5]byte),
+ error: "rlp: input string too long for [5]uint8",
+ },
+ {
+ input: "850101",
+ ptr: new([5]byte),
+ error: io.ErrUnexpectedEOF.Error(),
+ },
// byte array reuse (should be zeroed)
{input: "850102030405", ptr: &sharedByteArray, value: [5]byte{1, 2, 3, 4, 5}},
@@ -272,13 +290,23 @@ var decodeTests = []decodeTest{
{input: "C0", ptr: new(simplestruct), value: simplestruct{0, ""}},
{input: "C105", ptr: new(simplestruct), value: simplestruct{5, ""}},
{input: "C50583343434", ptr: new(simplestruct), value: simplestruct{5, "444"}},
- {input: "C3010101", ptr: new(simplestruct), error: "rlp: input list has too many elements for rlp.simplestruct"},
{
input: "C501C302C103",
ptr: new(recstruct),
value: recstruct{1, &recstruct{2, &recstruct{3, nil}}},
},
+ {
+ input: "C3010101",
+ ptr: new(simplestruct),
+ error: "rlp: input list has too many elements for rlp.simplestruct",
+ },
+ {
+ input: "C501C3C00000",
+ ptr: new(recstruct),
+ error: "rlp: expected input string or byte for uint, decoding into (rlp.recstruct).Child.I",
+ },
+
// pointers
{input: "00", ptr: new(*uint), value: (*uint)(nil)},
{input: "80", ptr: new(*uint), value: (*uint)(nil)},