diff --git a/cmd/ethereum/flags.go b/cmd/ethereum/flags.go index d27b739c36..275fcf248c 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 b238522820..8b83bbd376 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 fa284e52d0..ff6c21aa97 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 b59044bfce..7e19d003f5 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 db8e8e029e..065a4f7d84 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 7cfbc63f86..519c9fc13c 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 := §ion{} - 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 := §ion{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 = §ion{ + 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 := §ion{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 := §ion{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 315cc748db..b50a314ead 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 d1daad5750..1d9f806380 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 7c5d09489b..b67e5aaeab 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 322aec7b70..ab2aa289f0 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 ðProtocolTester{ + 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 0000000000..65728efa5b --- /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 0000000000..3da038be8b --- /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 0000000000..ad3c05b24a Binary files /dev/null and b/eth/test/chains/00.chain differ diff --git a/eth/test/chains/01.chain b/eth/test/chains/01.chain new file mode 100755 index 0000000000..56c9aef65f Binary files /dev/null and b/eth/test/chains/01.chain differ diff --git a/eth/test/chains/02.chain b/eth/test/chains/02.chain new file mode 100755 index 0000000000..440c92d658 Binary files /dev/null and b/eth/test/chains/02.chain differ diff --git a/eth/test/chains/03.chain b/eth/test/chains/03.chain new file mode 100755 index 0000000000..1cc7570ab4 Binary files /dev/null and b/eth/test/chains/03.chain differ diff --git a/eth/test/chains/04.chain b/eth/test/chains/04.chain new file mode 100755 index 0000000000..d4e5b1aa81 Binary files /dev/null and b/eth/test/chains/04.chain differ diff --git a/eth/test/mine.sh b/eth/test/mine.sh new file mode 100644 index 0000000000..0d95db1e43 --- /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 0000000000..5229af0353 --- /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 0000000000..9655cb3df7 --- /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 0000000000..9c5077164b --- /dev/null +++ b/eth/test/tests/00.sh @@ -0,0 +1,13 @@ +#!/bin/bash + +TIMEOUT=4 + +cat >> $JSFILE <> $JSFILE <> $JSFILE <> $JSFILE <> $JSFILE <> $JSFILE < 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 65f26fb12d..ce25b3e1b5 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 3267812343..cfff442f71 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 5c0d08d398..ceb89e3f7f 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 712d9fcf18..a2bd042859 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 7a1743937d..18ea63a090 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)},