diff options
Diffstat (limited to 'eth')
29 files changed, 3542 insertions, 0 deletions
diff --git a/eth/backend.go b/eth/backend.go new file mode 100644 index 000000000..b57e9fd69 --- /dev/null +++ b/eth/backend.go @@ -0,0 +1,309 @@ +package eth + +import ( + "crypto/ecdsa" + "fmt" + "strings" + + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/ethutil" + "github.com/ethereum/go-ethereum/event" + ethlogger "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/miner" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/nat" + "github.com/ethereum/go-ethereum/pow/ezp" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/whisper" +) + +var ( + logger = ethlogger.NewLogger("SERV") + jsonlogger = ethlogger.NewJsonLogger() + + defaultBootNodes = []*discover.Node{ + discover.MustParseNode("enode://6cdd090303f394a1cac34ecc9f7cda18127eafa2a3a06de39f6d920b0e583e062a7362097c7c65ee490a758b442acd5c80c6fce4b148c6a391e946b45131365b@54.169.166.226:30303"), + } +) + +type Config struct { + Name string + KeyStore string + DataDir string + LogFile string + LogLevel int + KeyRing string + LogFormat string + + MaxPeers int + Port string + + // This should be a space-separated list of + // discovery node URLs. + BootNodes string + + // This key is used to identify the node on the network. + // If nil, an ephemeral key is used. + NodeKey *ecdsa.PrivateKey + + NAT nat.Interface + Shh bool + Dial bool + + MinerThreads int + + KeyManager *crypto.KeyManager +} + +func (cfg *Config) parseBootNodes() []*discover.Node { + if cfg.BootNodes == "" { + return defaultBootNodes + } + var ns []*discover.Node + for _, url := range strings.Split(cfg.BootNodes, " ") { + if url == "" { + continue + } + n, err := discover.ParseNode(url) + if err != nil { + logger.Errorf("Bootstrap URL %s: %v\n", url, err) + continue + } + ns = append(ns, n) + } + return ns +} + +type Ethereum struct { + // Channel for shutting down the ethereum + shutdownChan chan bool + quit chan bool + + // DB interface + db ethutil.Database + blacklist p2p.Blacklist + + //*** SERVICES *** + // State manager for processing new blocks and managing the over all states + blockProcessor *core.BlockProcessor + txPool *core.TxPool + chainManager *core.ChainManager + blockPool *BlockPool + whisper *whisper.Whisper + + net *p2p.Server + eventMux *event.TypeMux + txSub event.Subscription + blockSub event.Subscription + miner *miner.Miner + + RpcServer rpc.RpcServer + WsServer rpc.RpcServer + keyManager *crypto.KeyManager + + logger ethlogger.LogSystem + + Mining bool +} + +func New(config *Config) (*Ethereum, error) { + // Boostrap database + logger := ethlogger.New(config.DataDir, config.LogFile, config.LogLevel, config.LogFormat) + db, err := ethdb.NewLDBDatabase("blockchain") + if err != nil { + return nil, err + } + + // Perform database sanity checks + d, _ := db.Get([]byte("ProtocolVersion")) + protov := ethutil.NewValue(d).Uint() + if protov != ProtocolVersion && protov != 0 { + return nil, fmt.Errorf("Database version mismatch. Protocol(%d / %d). `rm -rf %s`", protov, ProtocolVersion, ethutil.Config.ExecPath+"/database") + } + + // Create new keymanager + var keyManager *crypto.KeyManager + switch config.KeyStore { + case "db": + keyManager = crypto.NewDBKeyManager(db) + case "file": + keyManager = crypto.NewFileKeyManager(config.DataDir) + default: + return nil, fmt.Errorf("unknown keystore type: %s", config.KeyStore) + } + // Initialise the keyring + keyManager.Init(config.KeyRing, 0, false) + + saveProtocolVersion(db) + //ethutil.Config.Db = db + + eth := &Ethereum{ + shutdownChan: make(chan bool), + quit: make(chan bool), + db: db, + keyManager: keyManager, + blacklist: p2p.NewBlacklist(), + eventMux: &event.TypeMux{}, + logger: logger, + } + + eth.chainManager = core.NewChainManager(db, eth.EventMux()) + eth.txPool = core.NewTxPool(eth.EventMux()) + eth.blockProcessor = core.NewBlockProcessor(db, eth.txPool, eth.chainManager, eth.EventMux()) + eth.chainManager.SetProcessor(eth.blockProcessor) + eth.whisper = whisper.New() + eth.miner = miner.New(keyManager.Address(), eth, config.MinerThreads) + + hasBlock := eth.chainManager.HasBlock + insertChain := eth.chainManager.InsertChain + eth.blockPool = NewBlockPool(hasBlock, insertChain, ezp.Verify) + + ethProto := EthProtocol(eth.txPool, eth.chainManager, eth.blockPool) + protocols := []p2p.Protocol{ethProto} + if config.Shh { + protocols = append(protocols, eth.whisper.Protocol()) + } + + netprv := config.NodeKey + if netprv == nil { + if netprv, err = crypto.GenerateKey(); err != nil { + return nil, fmt.Errorf("could not generate server key: %v", err) + } + } + eth.net = &p2p.Server{ + PrivateKey: netprv, + Name: config.Name, + MaxPeers: config.MaxPeers, + Protocols: protocols, + Blacklist: eth.blacklist, + NAT: config.NAT, + NoDial: !config.Dial, + BootstrapNodes: config.parseBootNodes(), + } + if len(config.Port) > 0 { + eth.net.ListenAddr = ":" + config.Port + } + + return eth, nil +} + +func (s *Ethereum) KeyManager() *crypto.KeyManager { return s.keyManager } +func (s *Ethereum) Logger() ethlogger.LogSystem { return s.logger } +func (s *Ethereum) Name() string { return s.net.Name } +func (s *Ethereum) ChainManager() *core.ChainManager { return s.chainManager } +func (s *Ethereum) BlockProcessor() *core.BlockProcessor { return s.blockProcessor } +func (s *Ethereum) TxPool() *core.TxPool { return s.txPool } +func (s *Ethereum) BlockPool() *BlockPool { return s.blockPool } +func (s *Ethereum) Whisper() *whisper.Whisper { return s.whisper } +func (s *Ethereum) EventMux() *event.TypeMux { return s.eventMux } +func (s *Ethereum) Db() ethutil.Database { return s.db } +func (s *Ethereum) Miner() *miner.Miner { return s.miner } +func (s *Ethereum) IsListening() bool { return true } // Always listening +func (s *Ethereum) PeerCount() int { return s.net.PeerCount() } +func (s *Ethereum) Peers() []*p2p.Peer { return s.net.Peers() } +func (s *Ethereum) MaxPeers() int { return s.net.MaxPeers } +func (s *Ethereum) Coinbase() []byte { return nil } // TODO + +// Start the ethereum +func (s *Ethereum) Start() error { + jsonlogger.LogJson(ðlogger.LogStarting{ + ClientString: s.net.Name, + ProtocolVersion: ProtocolVersion, + }) + + err := s.net.Start() + if err != nil { + return err + } + + // Start services + s.txPool.Start() + s.blockPool.Start() + + if s.whisper != nil { + s.whisper.Start() + } + + // broadcast transactions + s.txSub = s.eventMux.Subscribe(core.TxPreEvent{}) + go s.txBroadcastLoop() + + // broadcast mined blocks + s.blockSub = s.eventMux.Subscribe(core.NewMinedBlockEvent{}) + go s.blockBroadcastLoop() + + logger.Infoln("Server started") + return nil +} + +func (self *Ethereum) SuggestPeer(nodeURL string) error { + n, err := discover.ParseNode(nodeURL) + if err != nil { + return fmt.Errorf("invalid node URL: %v", err) + } + self.net.SuggestPeer(n) + return nil +} + +func (s *Ethereum) Stop() { + // Close the database + defer s.db.Close() + + close(s.quit) + + s.txSub.Unsubscribe() // quits txBroadcastLoop + s.blockSub.Unsubscribe() // quits blockBroadcastLoop + + if s.RpcServer != nil { + s.RpcServer.Stop() + } + if s.WsServer != nil { + s.WsServer.Stop() + } + s.txPool.Stop() + s.eventMux.Stop() + s.blockPool.Stop() + if s.whisper != nil { + s.whisper.Stop() + } + + logger.Infoln("Server stopped") + close(s.shutdownChan) +} + +// This function will wait for a shutdown and resumes main thread execution +func (s *Ethereum) WaitForShutdown() { + <-s.shutdownChan +} + +// now tx broadcasting is taken out of txPool +// handled here via subscription, efficiency? +func (self *Ethereum) txBroadcastLoop() { + // automatically stops if unsubscribe + for obj := range self.txSub.Chan() { + event := obj.(core.TxPreEvent) + self.net.Broadcast("eth", TxMsg, event.Tx.RlpData()) + } +} + +func (self *Ethereum) blockBroadcastLoop() { + // automatically stops if unsubscribe + for obj := range self.blockSub.Chan() { + switch ev := obj.(type) { + case core.NewMinedBlockEvent: + self.net.Broadcast("eth", NewBlockMsg, ev.Block.RlpData(), ev.Block.Td) + } + } +} + +func saveProtocolVersion(db ethutil.Database) { + d, _ := db.Get([]byte("ProtocolVersion")) + protocolVersion := ethutil.NewValue(d).Uint() + + if protocolVersion == 0 { + db.Put([]byte("ProtocolVersion"), ethutil.NewValue(ProtocolVersion).Bytes()) + } +} diff --git a/eth/block_pool.go b/eth/block_pool.go new file mode 100644 index 000000000..13016c694 --- /dev/null +++ b/eth/block_pool.go @@ -0,0 +1,1239 @@ +package eth + +import ( + "bytes" + "fmt" + "math" + "math/big" + "math/rand" + "sort" + "sync" + "time" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethutil" + ethlogger "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/pow" +) + +var poolLogger = ethlogger.NewLogger("Blockpool") + +const ( + blockHashesBatchSize = 256 + blockBatchSize = 64 + blocksRequestInterval = 500 // ms + blocksRequestRepetition = 1 + blockHashesRequestInterval = 500 // ms + blocksRequestMaxIdleRounds = 100 + blockHashesTimeout = 60 // seconds + blocksTimeout = 120 // seconds +) + +type poolNode struct { + 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 + 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 + hasBlock func(hash []byte) bool + insertChain func(types.Blocks) error + verifyPoW func(pow.Block) bool +} + +type peerInfo struct { + lock sync.RWMutex + + td *big.Int + currentBlockHash []byte + currentBlock *types.Block + currentBlockC chan *types.Block + parentHash []byte + headSection *section + headSectionC chan *section + id string + + requestBlockHashes func([]byte) error + requestBlocks func([][]byte) error + peerError func(int, string, ...interface{}) + + sections map[string]*section + + 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, +) *BlockPool { + return &BlockPool{ + hasBlock: hasBlock, + insertChain: insertChain, + verifyPoW: verifyPoW, + } +} + +// allows restart +func (self *BlockPool) Start() { + self.lock.Lock() + if self.running { + self.lock.Unlock() + return + } + self.running = true + self.quit = make(chan bool) + self.flushC = make(chan bool) + self.pool = make(map[string]*poolEntry) + + self.lock.Unlock() + + self.peersLock.Lock() + self.peers = make(map[string]*peerInfo) + self.peersLock.Unlock() + + poolLogger.Infoln("Started") + +} + +func (self *BlockPool) Stop() { + self.lock.Lock() + if !self.running { + self.lock.Unlock() + return + } + self.running = false + + self.lock.Unlock() + + poolLogger.Infoln("Stopping...") + + close(self.quit) + //self.wg.Wait() + + self.peersLock.Lock() + self.peers = nil + self.peer = nil + 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, currentBlockHash []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) (best bool) { + + self.peersLock.Lock() + defer self.peersLock.Unlock() + peer, ok := self.peers[peerId] + if ok { + if bytes.Compare(peer.currentBlockHash, currentBlockHash) != 0 { + poolLogger.Debugf("Update peer %v with td %v and current block %s", peerId, td, name(currentBlockHash)) + peer.lock.Lock() + peer.td = td + peer.currentBlockHash = currentBlockHash + peer.currentBlock = nil + peer.parentHash = nil + peer.headSection = nil + peer.lock.Unlock() + } + } else { + peer = &peerInfo{ + td: td, + currentBlockHash: currentBlockHash, + id: peerId, //peer.Identity().Pubkey() + requestBlockHashes: requestBlockHashes, + requestBlocks: requestBlocks, + peerError: peerError, + sections: make(map[string]*section), + currentBlockC: make(chan *types.Block), + headSectionC: make(chan *section), + } + self.peers[peerId] = peer + poolLogger.Debugf("add new peer %v with td %v and current block %x", peerId, td, currentBlockHash[:4]) + } + // check peer current head + if self.hasBlock(currentBlockHash) { + // 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 new head section info from %s", peerId, name(currentBlockHash)) + peer.headSectionC <- nil + best = true + } else { + currentTD := ethutil.Big0 + if self.peer != nil { + currentTD = self.peer.td + } + if td.Cmp(currentTD) > 0 { + poolLogger.Debugf("peer %v promoted best peer", peerId) + self.switchPeer(self.peer, peer) + self.peer = peer + best = true + } + } + return +} + +func (self *BlockPool) requestHeadSection(peer *peerInfo) { + self.wg.Add(1) + self.procWg.Add(1) + poolLogger.Debugf("[%s] head section at [%s] requesting info", peer.id, name(peer.currentBlockHash)) + + go func() { + var idle bool + peer.lock.RLock() + quitC := peer.quitC + currentBlockHash := peer.currentBlockHash + peer.lock.RUnlock() + blockHashesRequestTimer := time.NewTimer(0) + blocksRequestTimer := time.NewTimer(0) + suicide := time.NewTimer(blockHashesTimeout * time.Second) + blockHashesRequestTimer.Stop() + defer blockHashesRequestTimer.Stop() + defer blocksRequestTimer.Stop() + + entry := self.get(currentBlockHash) + if entry != nil { + entry.node.lock.RLock() + currentBlock := entry.node.block + entry.node.lock.RUnlock() + if currentBlock != nil { + peer.lock.Lock() + peer.currentBlock = currentBlock + peer.parentHash = currentBlock.ParentHash() + poolLogger.Debugf("[%s] head block [%s] found", peer.id, name(currentBlockHash)) + peer.lock.Unlock() + blockHashesRequestTimer.Reset(0) + blocksRequestTimer.Stop() + } + } + + LOOP: + for { + + select { + case <-self.quit: + break LOOP + + case <-quitC: + poolLogger.Debugf("[%s] head section at [%s] incomplete - quit request loop", peer.id, name(currentBlockHash)) + break LOOP + + case headSection := <-peer.headSectionC: + peer.lock.Lock() + peer.headSection = headSection + if headSection == nil { + oldBlockHash := currentBlockHash + currentBlockHash = peer.currentBlockHash + poolLogger.Debugf("[%s] head section changed [%s] -> [%s]", peer.id, name(oldBlockHash), name(currentBlockHash)) + if idle { + idle = false + suicide.Reset(blockHashesTimeout * time.Second) + self.procWg.Add(1) + } + blocksRequestTimer.Reset(blocksRequestInterval * time.Millisecond) + } else { + poolLogger.DebugDetailf("[%s] head section at [%s] created", peer.id, name(currentBlockHash)) + if !idle { + idle = true + suicide.Stop() + self.procWg.Done() + } + } + peer.lock.Unlock() + blockHashesRequestTimer.Stop() + + case <-blockHashesRequestTimer.C: + poolLogger.DebugDetailf("[%s] head section at [%s] not found, requesting block hashes", peer.id, name(currentBlockHash)) + peer.requestBlockHashes(currentBlockHash) + blockHashesRequestTimer.Reset(blockHashesRequestInterval * time.Millisecond) + + case currentBlock := <-peer.currentBlockC: + peer.lock.Lock() + peer.currentBlock = currentBlock + peer.parentHash = currentBlock.ParentHash() + poolLogger.DebugDetailf("[%s] head block [%s] found", peer.id, name(currentBlockHash)) + peer.lock.Unlock() + if self.hasBlock(currentBlock.ParentHash()) { + if err := self.insertChain(types.Blocks([]*types.Block{currentBlock})); err != nil { + peer.peerError(ErrInvalidBlock, "%v", err) + } + if !idle { + idle = true + suicide.Stop() + self.procWg.Done() + } + } else { + blockHashesRequestTimer.Reset(0) + } + blocksRequestTimer.Stop() + + case <-blocksRequestTimer.C: + peer.lock.RLock() + poolLogger.DebugDetailf("[%s] head block [%s] not found, requesting", peer.id, name(currentBlockHash)) + peer.requestBlocks([][]byte{peer.currentBlockHash}) + peer.lock.RUnlock() + blocksRequestTimer.Reset(blocksRequestInterval * time.Millisecond) + + case <-suicide.C: + peer.peerError(ErrInsufficientChainInfo, "peer failed to provide block hashes or head block for block hash %x", currentBlockHash) + break LOOP + } + } + self.wg.Done() + if !idle { + self.procWg.Done() + } + }() +} + +// RemovePeer is called by the eth protocol when the peer disconnects +func (self *BlockPool) RemovePeer(peerId string) { + self.peersLock.Lock() + defer self.peersLock.Unlock() + peer, ok := self.peers[peerId] + if !ok { + return + } + delete(self.peers, peerId) + poolLogger.Debugf("remove peer %v", peerId) + + // if current best peer is removed, need find a better one + if self.peer == peer { + var newPeer *peerInfo + max := ethutil.Big0 + // peer with the highest self-acclaimed TD is chosen + for _, info := range self.peers { + if info.td.Cmp(max) > 0 { + max = info.td + newPeer = info + } + } + if newPeer != nil { + poolLogger.Debugf("peer %v with td %v promoted to best peer", newPeer.id, newPeer.td) + } else { + poolLogger.Warnln("no peers") + } + self.peer = newPeer + self.switchPeer(peer, newPeer) + } +} + +// Entry point for eth protocol to add block hashes received via BlockHashesMsg +// only hashes from the best peer is handled +// this method is always responsible to initiate further hash requests until +// a known parent is reached unless cancelled by a peerChange event +// this process also launches all request processes on each chain section +// this function needs to run asynchronously for one peer since the message is discarded??? +func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) { + + // register with peer manager loop + + peer, best := self.getPeer(peerId) + if !best { + return + } + // peer is still the best + + var size, n int + var hash []byte + var ok, headSection bool + var sec, child, parent *section + var entry *poolEntry + var nodes []*poolNode + bestPeer := peer + + hash, ok = next() + peer.lock.Lock() + if bytes.Compare(peer.parentHash, hash) == 0 { + if self.hasBlock(peer.currentBlockHash) { + return + } + poolLogger.Debugf("adding hashes at chain head for best peer %s starting from [%s]", peerId, name(peer.currentBlockHash)) + headSection = true + + if entry := self.get(peer.currentBlockHash); entry == nil { + node := &poolNode{ + hash: peer.currentBlockHash, + block: peer.currentBlock, + peer: peerId, + blockBy: peerId, + } + if size == 0 { + sec = newSection() + } + nodes = append(nodes, node) + size++ + n++ + } else { + child = entry.section + } + } else { + poolLogger.Debugf("adding hashes for best peer %s starting from [%s]", peerId, name(hash)) + } + quitC := peer.quitC + peer.lock.Unlock() + +LOOP: + // iterate using next (rlp stream lazy decoder) feeding hashesC + for ; ok; hash, ok = next() { + n++ + select { + case <-self.quit: + return + case <-quitC: + // if the peer is demoted, no more hashes taken + bestPeer = 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 sec != nil { + self.connectToBlockChain(sec) + } else { + if child != nil { + self.connectToBlockChain(child) + } + } + break LOOP + } + // 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 + poolLogger.DebugDetailf("[%s] connecting child section", sectionName(entry.section)) + 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 { + sec = 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(sec, nodes) + poolLogger.DebugDetailf("[%s]->[%s](%v)->[%s] new chain section", sectionName(parent), sectionName(sec), size, sectionName(child)) + self.link(parent, sec) + self.link(sec, child) + } else { + poolLogger.DebugDetailf("[%s]->[%s] connecting known sections", sectionName(parent), sectionName(child)) + self.link(parent, child) + } + + self.chainLock.Unlock() + + if parent != nil && bestPeer != nil { + self.activateChain(parent, peer) + poolLogger.Debugf("[%s] activate parent section [%s]", name(parent.top.hash), sectionName(parent)) + } + + if sec != nil { + peer.addSection(sec.top.hash, sec) + // request next section here once, only repeat if bottom block arrives, + // otherwise no way to check if it arrived + peer.requestBlockHashes(sec.bottom.hash) + sec.controlC <- bestPeer + poolLogger.Debugf("[%s] activate new section", sectionName(sec)) + } + + if headSection { + var headSec *section + switch { + case sec != nil: + headSec = sec + case child != nil: + headSec = child + default: + headSec = parent + } + peer.headSectionC <- headSec + } +} + +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 +// It has a strict interpretation of the protocol in that if the block received has not been requested, it results in an error (which can be ignored) +// block is checked for PoW +// only the first PoW-valid block for a hash is considered legit +func (self *BlockPool) AddBlock(block *types.Block, peerId string) { + hash := block.Hash() + self.peersLock.Lock() + peer := self.peer + self.peersLock.Unlock() + + entry := self.get(hash) + if bytes.Compare(hash, peer.currentBlockHash) == 0 { + poolLogger.Debugf("add head block [%s] for peer %s", name(hash), peerId) + peer.currentBlockC <- block + } else { + if entry == nil { + poolLogger.Warnf("unrequested block [%s] by peer %s", name(hash), peerId) + self.peerError(peerId, ErrUnrequestedBlock, "%x", hash) + } + } + if entry == nil { + return + } + + node := entry.node + node.lock.Lock() + defer node.lock.Unlock() + + // check if block already present + if node.block != nil { + poolLogger.DebugDetailf("block [%s] already sent by %s", name(hash), node.blockBy) + return + } + + if self.hasBlock(hash) { + poolLogger.DebugDetailf("block [%s] already known", name(hash)) + } else { + + // validate block for PoW + if !self.verifyPoW(block) { + poolLogger.Warnf("invalid pow on block [%s %v] by peer %s", name(hash), block.Number(), peerId) + self.peerError(peerId, ErrInvalidPoW, "%x", hash) + return + } + } + poolLogger.DebugDetailf("added block [%s] sent by peer %s", name(hash), peerId) + node.block = block + node.blockBy = peerId + +} + +func (self *BlockPool) connectToBlockChain(section *section) { + select { + case <-section.offC: + self.addSectionToBlockChain(section) + case <-section.blockChainC: + default: + close(section.blockChainC) + } +} + +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 + } + 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 +// - kills the section if there are blocks missing after an absolute time +// - kills the section if there are maxIdleRounds of idle rounds of block requests with no response +// - periodically polls the chain section for missing blocks which are then requested from peers +// - registers the process controller on the peer so that if the peer is promoted as best peer the second time (after a disconnect of a better one), all active processes are switched back on unless they expire and killed () +// - when turned off (if peer disconnects and new peer connects with alternative chain), no blockrequests are made but absolute expiry timer is ticking +// - when turned back on it recursively calls itself on the root of the next chain section +// - when exits, signals to +func (self *BlockPool) processSection(sec *section, nodes []*poolNode) { + + for i, node := range nodes { + entry := &poolEntry{node: node, section: sec, index: i} + self.set(node.hash, entry) + } + + sec.bottom = nodes[len(nodes)-1] + sec.top = nodes[0] + sec.nodes = nodes + poolLogger.DebugDetailf("[%s] setup section process", sectionName(sec)) + + self.wg.Add(1) + go func() { + + // absolute time after which sub-chain is killed if not complete (some blocks are missing) + suicideTimer := time.After(blocksTimeout * time.Second) + + 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 = sec.blockChainC + + var parentHash []byte + + LOOP: + for { + + if insertChain { + insertChain = false + rest, err := self.addSectionToBlockChain(sec) + if err != nil { + close(sec.suicideC) + continue LOOP + } + if rest == 0 { + blocksRequestsComplete = true + child := self.getChild(sec) + if child != nil { + self.connectToBlockChain(child) + } + } + } + + 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(sec), 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(sec), blocksRequests, missing, lastMissing, depth) + blocksRequestsComplete = true + blocksRequestTimer = nil + blocksRequestTime = false + } else { + poolLogger.DebugDetailf("[%s] section checked: missing %v/%v/%v", sectionName(sec), missing, lastMissing, depth) + // 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(sec), idle, blocksRequests, missing, lastMissing, depth) + close(sec.suicideC) + } + } else { + idle = 0 + } + same = true + } else { + same = false + } + } + lastMissing = missing + ready = true + done = false + // save a new processC (blocks still missing) + offC = missingC + missingC = processC + // put processC offline + processC = nil + } + // + + if ready && blocksRequestTime && !blocksRequestsComplete { + poolLogger.DebugDetailf("[%s] check if new blocks arrived (attempt %v): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth) + blocksRequestTimer = time.After(blocksRequestInterval * time.Millisecond) + blocksRequestTime = false + processC = offC + } + + if blockHashesRequestTime { + var parentSection = self.getParent(sec) + if parentSection == nil { + if parent := self.get(parentHash); parent != nil { + parentSection = parent.section + self.chainLock.Lock() + self.link(parentSection, sec) + self.chainLock.Unlock() + } else { + if self.hasBlock(parentHash) { + insertChain = true + blockHashesRequestTime = false + blockHashesRequestTimer = nil + blockHashesRequestsComplete = true + continue LOOP + } + } + } + if parentSection != nil { + // if not root of chain, switch off + poolLogger.DebugDetailf("[%s] parent found, hash requests deactivated (after %v total attempts)\n", sectionName(sec), blockHashesRequests) + blockHashesRequestTimer = nil + blockHashesRequestsComplete = true + } else { + blockHashesRequests++ + poolLogger.Debugf("[%s] hash request on root (%v total attempts)\n", sectionName(sec), blockHashesRequests) + peer.requestBlockHashes(sec.bottom.hash) + blockHashesRequestTimer = time.After(blockHashesRequestInterval * time.Millisecond) + } + blockHashesRequestTime = false + } + + select { + case <-self.quit: + break LOOP + + case <-quitC: + // peer quit or demoted, put section in idle mode + quitC = nil + go func() { + sec.controlC <- nil + }() + + case <-self.purgeC: + suicideTimer = time.After(0) + + case <-suicideTimer: + close(sec.suicideC) + poolLogger.Debugf("[%s] timeout. (%v total attempts): missing %v/%v/%v", sectionName(sec), blocksRequests, missing, lastMissing, depth) + + case <-sec.suicideC: + poolLogger.Debugf("[%s] suicide", sectionName(sec)) + + // first delink from child and parent under chainlock + self.chainLock.Lock() + self.link(nil, sec) + self.link(sec, nil) + self.chainLock.Unlock() + // delete node entries from pool index under pool lock + self.lock.Lock() + for _, node := range sec.nodes { + delete(self.pool, string(node.hash)) + } + self.lock.Unlock() + + break LOOP + + case <-blocksRequestTimer: + poolLogger.DebugDetailf("[%s] block request time", sectionName(sec)) + blocksRequestTime = true + + case <-blockHashesRequestTimer: + poolLogger.DebugDetailf("[%s] hash request time", sectionName(sec)) + blockHashesRequestTime = true + + case newPeer = <-sec.controlC: + + // 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(sec), blocksRequests, missing, lastMissing, depth) + } + blocksRequestTime = false + blocksRequestTimer = nil + blockHashesRequestTime = false + blockHashesRequestTimer = nil + if processC != nil { + offC = processC + processC = nil + } + } + + // idle -> active + if peer == nil && newPeer != nil { + self.procWg.Add(1) + + poolLogger.Debugf("[%s] active mode", sectionName(sec)) + if !blocksRequestsComplete { + blocksRequestTime = true + } + if !blockHashesRequestsComplete && parentHash != nil { + blockHashesRequestTime = true + } + 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(sec.nodes) + lastMissing = depth + // if not run at least once fully, launch iterator + go func() { + var node *poolNode + IT: + for _, node = range sec.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(sec)) + processC = offC + } + } + // reset quitC to current best peer + if newPeer != nil { + quitC = newPeer.quitC + } + peer = newPeer + + case waiter := <-sec.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 && !init { + // channel closed, first iteration finished + init = true + done = true + processC = make(chan *poolNode, missing) + poolLogger.DebugDetailf("[%s] section initalised: missing %v/%v/%v", sectionName(sec), missing, lastMissing, depth) + continue LOOP + } + if ready { + i = 0 + missing = 0 + ready = false + } + i++ + // if node has no block + node.lock.RLock() + block := node.block + node.lock.RUnlock() + if block == nil { + missing++ + hashes = append(hashes, node.hash) + if len(hashes) == blockBatchSize { + poolLogger.Debugf("[%s] request %v missing blocks", sectionName(sec), len(hashes)) + self.requestBlocks(blocksRequests, hashes) + hashes = nil + } + missingC <- node + } else { + if i == lastMissing { + if blockChainC == nil { + insertChain = true + } else { + if parentHash == nil { + parentHash = block.ParentHash() + poolLogger.Debugf("[%s] found root block [%s]", sectionName(sec), name(parentHash)) + blockHashesRequestTime = true + } + } + } + } + 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(sec)) + blockChainC = nil + // switch off hash requests in case they were on + blockHashesRequestTime = false + blockHashesRequestTimer = nil + blockHashesRequestsComplete = true + // section root has block + if len(sec.nodes) > 0 && sec.nodes[len(sec.nodes)-1].block != nil { + insertChain = true + } + continue LOOP + + } // select + } // for + + close(sec.offC) + + self.wg.Done() + if peer != nil { + self.procWg.Done() + } + }() + return +} + +func (self *BlockPool) peerError(peerId string, code int, format string, params ...interface{}) { + self.peersLock.RLock() + defer self.peersLock.RUnlock() + peer, ok := self.peers[peerId] + if ok { + peer.peerError(code, format, params...) + } +} + +func (self *BlockPool) requestBlocks(attempts int, hashes [][]byte) { + 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 [%x/%x] from peer %s", len(hashes), hashes[0][:4], hashes[len(hashes)-1][:4], peer.id) + peer.requestBlocks(hashes) + indexes = indexes[1:] + if len(indexes) == 0 { + break + } + } + i++ + } + self.wg.Done() + self.procWg.Done() + }() +} + +func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) { + self.peersLock.RLock() + defer self.peersLock.RUnlock() + if self.peer != nil && self.peer.id == peerId { + return self.peer, true + } + info, ok := self.peers[peerId] + if !ok { + return nil, false + } + return info, false +} + +func (self *peerInfo) addSection(hash []byte, section *section) (found *section) { + self.lock.Lock() + defer self.lock.Unlock() + key := string(hash) + found = self.sections[key] + poolLogger.DebugDetailf("[%s] section process stored for %s", sectionName(section), self.id) + self.sections[key] = section + return +} + +func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) { + if newPeer != nil { + newPeer.quitC = make(chan bool) + poolLogger.DebugDetailf("[%s] activate section processes", newPeer.id) + var addSections []*section + for hash, section := range newPeer.sections { + // split sections get reorganised here + if string(section.top.hash) != hash { + addSections = append(addSections, section) + if entry := self.get([]byte(hash)); entry != nil { + addSections = append(addSections, entry.section) + } + } + } + for _, section := range addSections { + newPeer.sections[string(section.top.hash)] = section + } + 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] activates section [%s]", newPeer.id, hash[:4], sectionName(section)) + } + } + newPeer.lock.Lock() + headSection := newPeer.headSection + currentBlockHash := newPeer.currentBlockHash + newPeer.lock.Unlock() + if headSection == nil { + poolLogger.DebugDetailf("[%s] head section for [%s] not created, requesting info", newPeer.id, name(currentBlockHash)) + self.requestHeadSection(newPeer) + } else { + if entry := self.get(currentBlockHash); entry != nil { + headSection = entry.section + } + poolLogger.DebugDetailf("[%s] activate chain at head section [%s] for current head [%s]", newPeer.id, sectionName(headSection), name(currentBlockHash)) + self.activateChain(headSection, newPeer) + } + } + if oldPeer != nil { + poolLogger.DebugDetailf("[%s] quit section processes", oldPeer.id) + close(oldPeer.quitC) + } +} + +func (self *BlockPool) getParent(sec *section) *section { + self.chainLock.RLock() + defer self.chainLock.RUnlock() + return sec.parent +} + +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 +} + +// 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 exChild != nil && exChild != child { + poolLogger.Debugf("[%s] chain fork [%s] -> [%s]", sectionName(parent), sectionName(exChild), sectionName(child)) + exChild.parent = nil + } + } + if child != nil { + 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 + } +} + +func (self *BlockPool) get(hash []byte) (node *poolEntry) { + self.lock.RLock() + defer self.lock.RUnlock() + return self.pool[string(hash)] +} + +func (self *BlockPool) set(hash []byte, node *poolEntry) { + self.lock.Lock() + defer self.lock.Unlock() + self.pool[string(hash)] = node +} diff --git a/eth/block_pool_test.go b/eth/block_pool_test.go new file mode 100644 index 000000000..331dbe504 --- /dev/null +++ b/eth/block_pool_test.go @@ -0,0 +1,983 @@ +package eth + +import ( + "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" +) + +const waitTimeout = 60 // seconds + +var logsys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel)) + +var ini = false + +func logInit() { + if !ini { + ethlogger.AddLogSystem(logsys) + ini = true + } +} + +// test helpers +func arrayEq(a, b []int) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + 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 (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 { + hash = newHash(i) + self.intToHash[i] = hash + self.hashToInt[string(hash)] = i + } + hashes = append(hashes, hash) + } + return +} + +func (self *testHashPool) hashesToIndexes(hashes [][]byte) (indexes []int) { + self.lock.Lock() + defer self.lock.Unlock() + for _, hash := range hashes { + i, found := self.hashToInt[string(hash)] + if !found { + i = -1 + } + indexes = append(indexes, i) + } + return +} + +// 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 + blocksRequestsMap map[int]bool + peerErrors []int + blockPool *BlockPool + hashPool *testHashPool + lock sync.RWMutex + id string + td int + currentBlock int + t *testing.T +} + +// 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), + } +} + +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) + } + } + } +} + +// 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: + } + } +} + +// 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) { + 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) + i := 1 + 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) { + 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.waitBlocksRequests(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("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.waitBlocksRequests(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.waitBlocksRequests(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") + } + peer0.waitBlocksRequests(0) + + 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.waitBlocksRequests(3) + + blockPool.Stop() + +} + +func TestPeerWithKnownBlock(t *testing.T) { + logInit() + _, blockPool, blockPoolTester := newTestBlockPool(t) + blockPoolTester.refBlockChain[0] = nil + blockPoolTester.blockChain[0] = nil + blockPool.Start() + + peer0 := blockPoolTester.newPeer("0", 1, 0) + peer0.AddPeer() + + blockPool.Wait(waitTimeout * time.Second) + blockPool.Stop() + // no request on known block + peer0.checkBlockHashesRequests() +} + +func TestPeerWithKnownParentBlock(t *testing.T) { + logInit() + _, blockPool, blockPoolTester := newTestBlockPool(t) + blockPoolTester.initRefBlockChain(1) + blockPoolTester.blockChain[0] = nil + blockPool.Start() + + peer0 := blockPoolTester.newPeer("0", 1, 1) + peer0.AddPeer() + peer0.AddBlocks(0, 1) + + blockPool.Wait(waitTimeout * time.Second) + blockPool.Stop() + peer0.checkBlocksRequests([]int{1}) + peer0.checkBlockHashesRequests() + blockPoolTester.refBlockChain[1] = []int{} + blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain) +} + +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() + peer1.AddBlocks(1, 2) + go peer1.AddBlockHashes(2, 1, 0) + peer1.AddBlocks(0, 1) + + blockPool.Wait(waitTimeout * time.Second) + blockPool.Stop() + blockPoolTester.refBlockChain[2] = []int{} + blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain) +} + +func TestChainConnectingWithParentHash(t *testing.T) { + logInit() + _, blockPool, blockPoolTester := newTestBlockPool(t) + blockPoolTester.blockChain[0] = nil + blockPoolTester.initRefBlockChain(3) + + blockPool.Start() + + peer1 := blockPoolTester.newPeer("peer1", 1, 3) + peer1.AddPeer() + go peer1.AddBlocks(2, 3) + go peer1.AddBlockHashes(3, 2, 1) + peer1.AddBlocks(0, 1, 2) + + blockPool.Wait(waitTimeout * time.Second) + blockPool.Stop() + blockPoolTester.refBlockChain[3] = []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.AddBlocks(2, 3) + go peer1.AddBlockHashes(3, 2, 1, 0) + peer1.AddBlocks(0, 1, 2) + + 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 %v", peer1.peerErrors) + } +} + +func TestVerifyPoW(t *testing.T) { + t.Skip("***FIX*** This test is broken") + 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] == 2 && !first { + first = true + return false + } else { + return true + } + + } + + blockPool.Start() + + peer1 := blockPoolTester.newPeer("peer1", 1, 3) + peer1.AddPeer() + go peer1.AddBlocks(2, 3) + go peer1.AddBlockHashes(3, 2, 1, 0) + peer1.AddBlocks(0, 1, 2) + + // blockPool.Wait(waitTimeout * time.Second) + time.Sleep(1 * time.Second) + blockPool.Stop() + blockPoolTester.refBlockChain[1] = []int{} + delete(blockPoolTester.refBlockChain, 2) + 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.AddBlocks(4, 5) + go peer1.AddBlockHashes(5, 4, 3) + go peer1.AddBlocks(2, 3, 4) + 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.AddBlocks(4, 5) // partially complete section + go peer1.AddBlockHashes(5, 4, 3) + peer1.AddBlocks(3, 4) // partially complete section + // peer1 found new blocks + peer1.td = 2 + peer1.currentBlock = 7 + peer1.AddPeer() + go peer1.AddBlocks(6, 7) + go peer1.AddBlockHashes(7, 6, 5) + go peer1.AddBlocks(2, 3) + go peer1.AddBlocks(5, 6) + 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 TestPeerSwitchUp(t *testing.T) { + logInit() + _, blockPool, blockPoolTester := newTestBlockPool(t) + blockPoolTester.blockChain[0] = nil + blockPoolTester.initRefBlockChain(7) + + blockPool.Start() + + peer1 := blockPoolTester.newPeer("peer1", 1, 6) + peer2 := blockPoolTester.newPeer("peer2", 2, 7) + peer2.blocksRequestsMap = peer1.blocksRequestsMap + + peer1.AddPeer() + go peer1.AddBlocks(5, 6) + go peer1.AddBlockHashes(6, 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.AddBlocks(6, 7) + go peer2.AddBlockHashes(7, 6) // + go peer2.AddBlocks(4, 5) // 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[7] = []int{} + blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain) +} + +func TestPeerSwitchDown(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() + peer2.AddBlocks(5, 6) // partially complete, section will be preserved + go peer2.AddBlockHashes(6, 5, 4) // + peer2.AddBlocks(4, 5) // + 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) // tests that section set by demoted peer is remembered and blocks are accepted , this connects the chain sections together + peer1.AddBlocks(0, 1, 2, 3) + + blockPool.Wait(waitTimeout * time.Second) + blockPool.Stop() + blockPoolTester.refBlockChain[6] = []int{} + blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain) +} + +func TestPeerCompleteSectionSwitchDown(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() + peer2.AddBlocks(5, 6) // partially complete, section will be preserved + go peer2.AddBlockHashes(6, 5, 4) // + peer2.AddBlocks(3, 4, 5) // complete section + blockPool.RemovePeer("peer2") // peer2 disconnects + peer1.AddPeer() // inferior peer1 is promoted as best peer + peer1.AddBlockHashes(4, 3, 2, 1, 0) // tests that hash request are directly connecting if the head block exists + 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.AddBlocks(7, 8) + 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.AddBlocks(10, 11) // + 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.AddBlocks(8, 9) + go peer1.AddBlockHashes(9, 8, 7, 3, 2) + peer1.AddBlocks(1, 2, 3, 7, 8) + peer2.AddPeer() // peer2 is promoted as best peer + go peer2.AddBlocks(5, 6) // + go peer2.AddBlockHashes(6, 5, 4, 3, 2) // fork on 3 -> 4 (earlier child: 7) + go peer2.AddBlocks(1, 2, 3, 4, 5) + 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() + peer1.AddBlocks(8, 9) // + go peer1.AddBlockHashes(9, 8, 7, 3, 2) // + peer1.AddBlocks(7, 8) // partial section + peer2.AddPeer() // + peer2.AddBlocks(5, 6) // + go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3 + peer2.AddBlocks(1, 2, 3, 4, 5) // + + // peer1 finds new blocks + peer1.td = 3 + peer1.currentBlock = 11 + peer1.AddPeer() + go peer1.AddBlocks(10, 11) + go peer1.AddBlockHashes(11, 10, 9) + peer1.AddBlocks(9, 10) + go peer1.AddBlocks(3, 7) // tests that block requests on earlier fork are 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) + + 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.AddBlocks(8, 9) + go peer1.AddBlockHashes(9, 8, 7, 3, 2) + peer1.AddBlocks(7, 8) + peer2.AddPeer() + go peer2.AddBlocks(5, 6) // + go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3 + peer2.AddBlocks(2, 3, 4, 5) // + blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer + go peer1.AddBlocks(3, 7) // tests that block requests on earlier fork are remembered and orphan section relinks to existing parent block + go peer1.AddBlocks(1, 2) // + go peer1.AddBlockHashes(2, 1, 0) // + peer1.AddBlocks(0, 1) + + 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.AddBlocks(8, 9) + go peer1.AddBlockHashes(9, 8, 7) + peer1.AddBlocks(3, 7, 8) // 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 block 2 missing + peer2.AddPeer() // + go peer2.AddBlocks(5, 6) // + go peer2.AddBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3 + peer2.AddBlocks(2, 3, 4, 5) // 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 new file mode 100644 index 000000000..9c4a68481 --- /dev/null +++ b/eth/error.go @@ -0,0 +1,72 @@ +package eth + +import ( + "fmt" +) + +const ( + ErrMsgTooLarge = iota + ErrDecode + ErrInvalidMsgCode + ErrProtocolVersionMismatch + ErrNetworkIdMismatch + ErrGenesisBlockMismatch + ErrNoStatusMsg + ErrExtraStatusMsg + ErrInvalidBlock + ErrInvalidPoW + ErrUnrequestedBlock + ErrInsufficientChainInfo +) + +var errorToString = map[int]string{ + ErrMsgTooLarge: "Message too long", + ErrDecode: "Invalid message", + ErrInvalidMsgCode: "Invalid message code", + ErrProtocolVersionMismatch: "Protocol version mismatch", + ErrNetworkIdMismatch: "NetworkId mismatch", + ErrGenesisBlockMismatch: "Genesis block mismatch", + ErrNoStatusMsg: "No status message", + ErrExtraStatusMsg: "Extra status message", + ErrInvalidBlock: "Invalid block", + ErrInvalidPoW: "Invalid PoW", + ErrUnrequestedBlock: "Unrequested block", + ErrInsufficientChainInfo: "Insufficient chain info", +} + +type protocolError struct { + Code int + fatal bool + message string + format string + params []interface{} + // size int +} + +func newProtocolError(code int, format string, params ...interface{}) *protocolError { + return &protocolError{Code: code, format: format, params: params} +} + +func ProtocolError(code int, format string, params ...interface{}) (err *protocolError) { + err = newProtocolError(code, format, params...) + // report(err) + return +} + +func (self protocolError) Error() (message string) { + if len(message) == 0 { + var ok bool + self.message, ok = errorToString[self.Code] + if !ok { + panic("invalid error code") + } + if self.format != "" { + self.message += ": " + fmt.Sprintf(self.format, self.params...) + } + } + return self.message +} + +func (self *protocolError) Fatal() bool { + return self.fatal +} diff --git a/eth/peer_util.go b/eth/peer_util.go new file mode 100644 index 000000000..6cf80cde2 --- /dev/null +++ b/eth/peer_util.go @@ -0,0 +1,23 @@ +package eth + +import ( + "encoding/json" + + "github.com/ethereum/go-ethereum/ethutil" +) + +func WritePeers(path string, addresses []string) { + if len(addresses) > 0 { + data, _ := json.MarshalIndent(addresses, "", " ") + ethutil.WriteFile(path, data) + } +} + +func ReadPeers(path string) (ips []string, err error) { + var data string + data, err = ethutil.ReadAllFile(path) + if err != nil { + json.Unmarshal([]byte(data), &ips) + } + return +} diff --git a/eth/protocol.go b/eth/protocol.go new file mode 100644 index 000000000..8221c1b29 --- /dev/null +++ b/eth/protocol.go @@ -0,0 +1,339 @@ +package eth + +import ( + "bytes" + "fmt" + "io" + "math/big" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethutil" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rlp" +) + +const ( + ProtocolVersion = 54 + NetworkId = 0 + ProtocolLength = uint64(8) + ProtocolMaxMsgSize = 10 * 1024 * 1024 +) + +// eth protocol message codes +const ( + StatusMsg = iota + GetTxMsg // unused + TxMsg + GetBlockHashesMsg + BlockHashesMsg + GetBlocksMsg + BlocksMsg + NewBlockMsg +) + +// ethProtocol represents the ethereum wire protocol +// instance is running on each peer +type ethProtocol struct { + txPool txPool + chainManager chainManager + blockPool blockPool + peer *p2p.Peer + id string + rw p2p.MsgReadWriter +} + +// backend is the interface the ethereum protocol backend should implement +// used as an argument to EthProtocol +type txPool interface { + AddTransactions([]*types.Transaction) + GetTransactions() types.Transactions +} + +type chainManager interface { + GetBlockHashesFromHash(hash []byte, amount uint64) (hashes [][]byte) + GetBlock(hash []byte) (block *types.Block) + Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) +} + +type blockPool interface { + AddBlockHashes(next func() ([]byte, bool), peerId string) + AddBlock(block *types.Block, peerId string) + AddPeer(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(int, string, ...interface{})) (best bool) + RemovePeer(peerId string) +} + +// message structs used for rlp decoding +type newBlockMsgData struct { + Block *types.Block + TD *big.Int +} + +const maxHashes = 255 + +type getBlockHashesMsgData struct { + Hash []byte + Amount uint64 +} + +// main entrypoint, wrappers starting a server running the eth protocol +// use this constructor to attach the protocol ("class") to server caps +// the Dev p2p layer then runs the protocol instance on each peer +func EthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool) p2p.Protocol { + return p2p.Protocol{ + Name: "eth", + Version: ProtocolVersion, + Length: ProtocolLength, + Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error { + return runEthProtocol(txPool, chainManager, blockPool, peer, rw) + }, + } +} + +// the main loop that handles incoming messages +// note RemovePeer in the post-disconnect hook +func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) { + id := peer.ID() + self := ðProtocol{ + txPool: txPool, + chainManager: chainManager, + blockPool: blockPool, + rw: rw, + peer: peer, + id: fmt.Sprintf("%x", id[:8]), + } + err = self.handleStatus() + if err == nil { + self.propagateTxs() + for { + err = self.handle() + if err != nil { + self.blockPool.RemovePeer(self.id) + break + } + } + } + return +} + +func (self *ethProtocol) handle() error { + msg, err := self.rw.ReadMsg() + if err != nil { + return err + } + if msg.Size > ProtocolMaxMsgSize { + return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + } + // make sure that the payload has been fully consumed + defer msg.Discard() + + switch msg.Code { + case GetTxMsg: // ignore + case StatusMsg: + return self.protoError(ErrExtraStatusMsg, "") + + case TxMsg: + // TODO: rework using lazy RLP stream + var txs []*types.Transaction + if err := msg.Decode(&txs); err != nil { + 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 self.protoError(ErrDecode, "->msg %v: %v", msg, err) + } + + //request.Amount = uint64(math.Min(float64(maxHashes), float64(request.Amount))) + if request.Amount > maxHashes { + request.Amount = maxHashes + } + hashes := self.chainManager.GetBlockHashesFromHash(request.Hash, request.Amount) + return p2p.EncodeMsg(self.rw, BlockHashesMsg, ethutil.ByteSliceToInterface(hashes)...) + + case BlockHashesMsg: + // TODO: redo using lazy decode , this way very inefficient on known chains + 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) + + case GetBlocksMsg: + msgStream := rlp.NewStream(msg.Payload) + var blocks []interface{} + 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) + } + if i == blockHashesBatchSize { + break + } + } + return p2p.EncodeMsg(self.rw, BlocksMsg, blocks...) + + case BlocksMsg: + msgStream := rlp.NewStream(msg.Payload) + for { + var block types.Block + if err := msgStream.Decode(&block); err != nil { + if err == io.EOF { + break + } else { + return self.protoError(ErrDecode, "msg %v: %v", msg, err) + } + } + self.blockPool.AddBlock(&block, self.id) + } + + case NewBlockMsg: + var request newBlockMsgData + if err := msg.Decode(&request); err != nil { + return self.protoError(ErrDecode, "msg %v: %v", msg, err) + } + hash := request.Block.Hash() + // to simplify backend interface adding a new block + // uses AddPeer followed by AddHashes, AddBlock only if peer is the best peer + // (or selected as new best peer) + if self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) { + self.blockPool.AddBlock(request.Block, self.id) + } + + default: + return self.protoError(ErrInvalidMsgCode, "%v", msg.Code) + } + return nil +} + +type statusMsgData struct { + ProtocolVersion uint32 + NetworkId uint32 + TD *big.Int + CurrentBlock []byte + GenesisBlock []byte +} + +func (self *ethProtocol) statusMsg() p2p.Msg { + td, currentBlock, genesisBlock := self.chainManager.Status() + + return p2p.NewMsg(StatusMsg, + uint32(ProtocolVersion), + uint32(NetworkId), + td, + currentBlock, + genesisBlock, + ) +} + +func (self *ethProtocol) handleStatus() error { + // send precanned status message + if err := self.rw.WriteMsg(self.statusMsg()); err != nil { + return err + } + + // read and handle remote status + msg, err := self.rw.ReadMsg() + if err != nil { + return err + } + + if msg.Code != StatusMsg { + return self.protoError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg) + } + + if msg.Size > ProtocolMaxMsgSize { + return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + } + + var status statusMsgData + if err := msg.Decode(&status); err != nil { + return self.protoError(ErrDecode, "msg %v: %v", msg, err) + } + + _, _, genesisBlock := self.chainManager.Status() + + if bytes.Compare(status.GenesisBlock, genesisBlock) != 0 { + return self.protoError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock) + } + + if status.NetworkId != NetworkId { + return self.protoError(ErrNetworkIdMismatch, "%d (!= %d)", status.NetworkId, NetworkId) + } + + if ProtocolVersion != status.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) + + return nil +} + +func (self *ethProtocol) requestBlockHashes(from []byte) error { + self.peer.Debugf("fetching hashes (%d) %x...\n", blockHashesBatchSize, from[0:4]) + return p2p.EncodeMsg(self.rw, GetBlockHashesMsg, interface{}(from), uint64(blockHashesBatchSize)) +} + +func (self *ethProtocol) requestBlocks(hashes [][]byte) error { + self.peer.Debugf("fetching %v blocks", len(hashes)) + return p2p.EncodeMsg(self.rw, 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 %v", err) + // disconnect + } else { + self.peer.Debugf("fyi %v", err) + } + return +} + +func (self *ethProtocol) protoErrorDisconnect(code int, format string, params ...interface{}) { + err := ProtocolError(code, format, params...) + if err.Fatal() { + self.peer.Errorln("err %v", err) + // disconnect + } else { + self.peer.Debugf("fyi %v", err) + } + +} + +func (self *ethProtocol) propagateTxs() { + transactions := self.txPool.GetTransactions() + iface := make([]interface{}, len(transactions)) + for i, transaction := range transactions { + iface[i] = transaction + } + + self.rw.WriteMsg(p2p.NewMsg(TxMsg, iface...)) +} diff --git a/eth/protocol_test.go b/eth/protocol_test.go new file mode 100644 index 000000000..a91806a1c --- /dev/null +++ b/eth/protocol_test.go @@ -0,0 +1,254 @@ +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" + "github.com/ethereum/go-ethereum/p2p/discover" +) + +var sys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel)) + +type testMsgReadWriter struct { + in chan p2p.Msg + out []p2p.Msg +} + +func (self *testMsgReadWriter) In(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 = append(self.out, msg) + return nil +} + +func (self *testMsgReadWriter) ReadMsg() (p2p.Msg, error) { + msg, ok := <-self.in + if !ok { + return msg, io.EOF + } + return msg, nil +} + +type testTxPool struct { + getTransactions func() []*types.Transaction + addTransactions func(txs []*types.Transaction) +} + +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) +} + +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 *testTxPool) GetTransactions() types.Transactions { return nil } + +func (self *testChainManager) GetBlockHashesFromHash(hash []byte, amount uint64) (hashes [][]byte) { + if self.getBlockHashes != nil { + hashes = self.getBlockHashes(hash, amount) + } + return +} + +func (self *testChainManager) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) { + if self.status != nil { + td, currentBlock, genesisBlock = self.status() + } + return +} + +func (self *testChainManager) GetBlock(hash []byte) (block *types.Block) { + if self.getBlock != nil { + block = self.getBlock(hash) + } + return +} + +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 { + self.addBlock(block, peerId) + } +} + +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 { + best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError) + } + return +} + +func (self *testBlockPool) RemovePeer(peerId string) { + if self.removePeer != nil { + self.removePeer(peerId) + } +} + +func testPeer() *p2p.Peer { + var id discover.NodeID + pk := crypto.GenerateNewKeyPair().PublicKey + copy(id[:], pk) + return p2p.NewPeer(id, "test peer", []p2p.Cap{}) +} + +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) + eth.In(statusMsg) + delay := 1 * time.Second + eth.checkError(ErrNoStatusMsg, delay) + var status statusMsgData + eth.checkMsg(0, StatusMsg, &status) // first outgoing msg should be StatusMsg + if status.TD.Cmp(td) != 0 || + status.ProtocolVersion != ProtocolVersion || + status.NetworkId != NetworkId || + status.TD.Cmp(td) != 0 || + bytes.Compare(status.CurrentBlock, currentBlock) != 0 || + bytes.Compare(status.GenesisBlock, genesis) != 0 { + t.Errorf("incorrect outgoing status") + } + + eth.reset() + go eth.run() + statusMsg = p2p.NewMsg(0, uint32(48), uint32(0), td, currentBlock, genesis) + eth.In(statusMsg) + eth.checkError(ErrProtocolVersionMismatch, delay) + + eth.reset() + go eth.run() + statusMsg = p2p.NewMsg(0, uint32(49), uint32(1), td, currentBlock, genesis) + eth.In(statusMsg) + eth.checkError(ErrNetworkIdMismatch, delay) + + eth.reset() + go eth.run() + statusMsg = p2p.NewMsg(0, uint32(49), uint32(0), td, currentBlock, []byte{3}) + eth.In(statusMsg) + eth.checkError(ErrGenesisBlockMismatch, delay) + +} diff --git a/eth/test/README.md b/eth/test/README.md new file mode 100644 index 000000000..65728efa5 --- /dev/null +++ b/eth/test/README.md @@ -0,0 +1,27 @@ += Integration tests for eth protocol and blockpool + +This is a simple suite of tests to fire up a local test node with peers to test blockchain synchronisation and download. +The scripts call ethereum (assumed to be compiled in go-ethereum root). + +To run a test: + + . run.sh 00 02 + +Without arguments, all tests are run. + +Peers are launched with preloaded imported chains. In order to prevent them from synchronizing with each other they are set with `-dial=false` and `-maxpeer 1` options. They log into `/tmp/eth.test/nodes/XX` where XX is the last two digits of their port. + +Chains to import can be bootstrapped by letting nodes mine for some time. This is done with + + . bootstrap.sh + +Only the relative timing and forks matter so they should work if the bootstrap script is rerun. +The reference blockchain of tests are soft links to these import chains and check at the end of a test run. + +Connecting to peers and exporting blockchain is scripted with JS files executed by the JSRE, see `tests/XX.sh`. + +Each test is set with a timeout. This may vary on different computers so adjust sensibly. +If you kill a test before it completes, do not forget to kill all the background processes, since they will impact the result. Use: + + killall ethereum + diff --git a/eth/test/bootstrap.sh b/eth/test/bootstrap.sh new file mode 100644 index 000000000..3da038be8 --- /dev/null +++ b/eth/test/bootstrap.sh @@ -0,0 +1,9 @@ +#!/bin/bash +# bootstrap chains - used to regenerate tests/chains/*.chain + +mkdir -p chains +bash ./mine.sh 00 10 +bash ./mine.sh 01 5 00 +bash ./mine.sh 02 10 00 +bash ./mine.sh 03 5 02 +bash ./mine.sh 04 10 02
\ No newline at end of file diff --git a/eth/test/chains/00.chain b/eth/test/chains/00.chain Binary files differnew file mode 100755 index 000000000..ad3c05b24 --- /dev/null +++ b/eth/test/chains/00.chain diff --git a/eth/test/chains/01.chain b/eth/test/chains/01.chain Binary files differnew file mode 100755 index 000000000..56c9aef65 --- /dev/null +++ b/eth/test/chains/01.chain diff --git a/eth/test/chains/02.chain b/eth/test/chains/02.chain Binary files differnew file mode 100755 index 000000000..440c92d65 --- /dev/null +++ b/eth/test/chains/02.chain diff --git a/eth/test/chains/03.chain b/eth/test/chains/03.chain Binary files differnew file mode 100755 index 000000000..1cc7570ab --- /dev/null +++ b/eth/test/chains/03.chain diff --git a/eth/test/chains/04.chain b/eth/test/chains/04.chain Binary files differnew file mode 100755 index 000000000..d4e5b1aa8 --- /dev/null +++ b/eth/test/chains/04.chain diff --git a/eth/test/mine.sh b/eth/test/mine.sh new file mode 100644 index 000000000..0d95db1e4 --- /dev/null +++ b/eth/test/mine.sh @@ -0,0 +1,20 @@ +#!/bin/bash +# bash ./mine.sh node_id timeout(sec) [basechain] +ETH=../../ethereum +MINE="$ETH -datadir tmp/nodes/$1 -seed=false -port '' -shh=false -id test$1" +rm -rf tmp/nodes/$1 +echo "Creating chain $1..." +if [[ "" != "$3" ]]; then + CHAIN="chains/$3.chain" + CHAINARG="-chain $CHAIN" + $MINE -mine $CHAINARG -loglevel 3 | grep 'importing' +fi +$MINE -mine -loglevel 0 & +PID=$! +sleep $2 +kill $PID +$MINE -loglevel 3 <(echo "eth.export(\"chains/$1.chain\")") > /tmp/eth.test/mine.tmp & +PID=$! +sleep 1 +kill $PID +cat /tmp/eth.test/mine.tmp | grep 'exporting' diff --git a/eth/test/run.sh b/eth/test/run.sh new file mode 100644 index 000000000..5229af035 --- /dev/null +++ b/eth/test/run.sh @@ -0,0 +1,53 @@ +#!/bin/bash +# bash run.sh (testid0 testid1 ...) +# runs tests tests/testid0.sh tests/testid1.sh ... +# without arguments, it runs all tests + +. tests/common.sh + +TESTS= + +if [ "$#" -eq 0 ]; then + for NAME in tests/??.sh; do + i=`basename $NAME .sh` + TESTS="$TESTS $i" + done +else + TESTS=$@ +fi + +ETH=../../ethereum +DIR="/tmp/eth.test/nodes" +TIMEOUT=10 + +mkdir -p $DIR/js + +echo "running tests $TESTS" +for NAME in $TESTS; do + PIDS= + CHAIN="tests/$NAME.chain" + JSFILE="$DIR/js/$NAME.js" + CHAIN_TEST="$DIR/$NAME/chain" + + echo "RUN: test $NAME" + cat tests/common.js > $JSFILE + . tests/$NAME.sh + sleep $TIMEOUT + echo "timeout after $TIMEOUT seconds: killing $PIDS" + kill $PIDS + if [ -r "$CHAIN" ]; then + if diff $CHAIN $CHAIN_TEST >/dev/null ; then + echo "chain ok: $CHAIN=$CHAIN_TEST" + else + echo "FAIL: chains differ: expected $CHAIN ; got $CHAIN_TEST" + continue + fi + fi + ERRORS=$DIR/errors + if [ -r "$ERRORS" ]; then + echo "FAIL: " + cat $ERRORS + else + echo PASS + fi +done
\ No newline at end of file diff --git a/eth/test/tests/00.chain b/eth/test/tests/00.chain new file mode 120000 index 000000000..9655cb3df --- /dev/null +++ b/eth/test/tests/00.chain @@ -0,0 +1 @@ +../chains/01.chain
\ No newline at end of file diff --git a/eth/test/tests/00.sh b/eth/test/tests/00.sh new file mode 100644 index 000000000..9c5077164 --- /dev/null +++ b/eth/test/tests/00.sh @@ -0,0 +1,13 @@ +#!/bin/bash + +TIMEOUT=4 + +cat >> $JSFILE <<EOF +eth.addPeer("localhost:30311"); +sleep(1000) +eth.export("$CHAIN_TEST"); +EOF + +peer 11 01 +test_node $NAME "" -loglevel 5 $JSFILE + diff --git a/eth/test/tests/01.chain b/eth/test/tests/01.chain new file mode 120000 index 000000000..ae65ccb37 --- /dev/null +++ b/eth/test/tests/01.chain @@ -0,0 +1 @@ +../chains/02.chain
\ No newline at end of file diff --git a/eth/test/tests/01.sh b/eth/test/tests/01.sh new file mode 100644 index 000000000..1601e0dfb --- /dev/null +++ b/eth/test/tests/01.sh @@ -0,0 +1,18 @@ +#!/bin/bash + +TIMEOUT=5 + +cat >> $JSFILE <<EOF +eth.addPeer("localhost:30311"); +log("added peer localhost:30311"); +sleep(1000); +log("added peer localhost:30312"); +eth.addPeer("localhost:30312"); +sleep(3000); +eth.export("$CHAIN_TEST"); +EOF + +peer 11 01 +peer 12 02 +test_node $NAME "" -loglevel 5 $JSFILE + diff --git a/eth/test/tests/02.chain b/eth/test/tests/02.chain new file mode 120000 index 000000000..9655cb3df --- /dev/null +++ b/eth/test/tests/02.chain @@ -0,0 +1 @@ +../chains/01.chain
\ No newline at end of file diff --git a/eth/test/tests/02.sh b/eth/test/tests/02.sh new file mode 100644 index 000000000..619217ed8 --- /dev/null +++ b/eth/test/tests/02.sh @@ -0,0 +1,19 @@ +#!/bin/bash + +TIMEOUT=6 + +cat >> $JSFILE <<EOF +eth.addPeer("localhost:30311"); +sleep(200); +eth.addPeer("localhost:30312"); +sleep(3000); +eth.export("$CHAIN_TEST"); +EOF + +peer 11 01 +peer 12 02 +P12ID=$PID +test_node $NAME "" -loglevel 5 $JSFILE +sleep 0.3 +kill $P12ID + diff --git a/eth/test/tests/03.chain b/eth/test/tests/03.chain new file mode 120000 index 000000000..b07c49a30 --- /dev/null +++ b/eth/test/tests/03.chain @@ -0,0 +1 @@ +../chains/12k.chain
\ No newline at end of file diff --git a/eth/test/tests/03.sh b/eth/test/tests/03.sh new file mode 100644 index 000000000..d7dba737f --- /dev/null +++ b/eth/test/tests/03.sh @@ -0,0 +1,14 @@ +#!/bin/bash + +TIMEOUT=12 + +cat >> $JSFILE <<EOF +eth.addPeer("localhost:30311"); +sleep(10000); +eth.export("$CHAIN_TEST"); +EOF + +peer 11 12k +sleep 2 +test_node $NAME "" -loglevel 5 $JSFILE + diff --git a/eth/test/tests/04.sh b/eth/test/tests/04.sh new file mode 100644 index 000000000..d77c360ba --- /dev/null +++ b/eth/test/tests/04.sh @@ -0,0 +1,17 @@ +#!/bin/bash + +TIMEOUT=15 + +cat >> $JSFILE <<EOF +eth.addPeer("localhost:30311"); +sleep(200); +eth.addPeer("localhost:30312"); +sleep(13000); +eth.export("$CHAIN_TEST"); +EOF + +peer 11 01 -mine +peer 12 02 +test_node $NAME "" -loglevel 5 $JSFILE +sleep 6 +cat $DIR/$NAME/debug.log | grep 'best peer' diff --git a/eth/test/tests/05.sh b/eth/test/tests/05.sh new file mode 100644 index 000000000..0a0f94df5 --- /dev/null +++ b/eth/test/tests/05.sh @@ -0,0 +1,20 @@ +#!/bin/bash + +TIMEOUT=60 + +cat >> $JSFILE <<EOF +eth.addPeer("localhost:30311"); +sleep(200); +eth.addPeer("localhost:30312"); +eth.addPeer("localhost:30313"); +eth.addPeer("localhost:30314"); +sleep(3000); +eth.export("$CHAIN_TEST"); +EOF + +peer 11 01 -mine +peer 12 02 -mine +peer 13 03 +peer 14 04 +test_node $NAME "" -loglevel 5 $JSFILE + diff --git a/eth/test/tests/common.js b/eth/test/tests/common.js new file mode 100644 index 000000000..206ebf145 --- /dev/null +++ b/eth/test/tests/common.js @@ -0,0 +1,9 @@ +function log(text) { + console.log("[JS TEST SCRIPT] " + text); +} + +function sleep(seconds) { + var now = new Date().getTime(); + while(new Date().getTime() < now + seconds){} +} + diff --git a/eth/test/tests/common.sh b/eth/test/tests/common.sh new file mode 100644 index 000000000..f66cfb43e --- /dev/null +++ b/eth/test/tests/common.sh @@ -0,0 +1,20 @@ +#!/bin/bash + +# launched by run.sh +function test_node { + rm -rf $DIR/$1 + ARGS="-datadir $DIR/$1 -debug debug -seed=false -shh=false -id test$1 -port 303$1" + if [ "" != "$2" ]; then + chain="chains/$2.chain" + echo "import chain $chain" + $ETH $ARGS -loglevel 3 -chain $chain | grep CLI |grep import + fi + echo "starting test node $1 with args $ARGS ${@:3}" + $ETH $ARGS ${@:3} & + PID=$! + PIDS="$PIDS $PID" +} + +function peer { + test_node $@ -loglevel 5 -logfile debug.log -maxpeer 1 -dial=false +}
\ No newline at end of file diff --git a/eth/wallet.go b/eth/wallet.go new file mode 100644 index 000000000..9ec834309 --- /dev/null +++ b/eth/wallet.go @@ -0,0 +1,80 @@ +package eth + +/* +import ( + "crypto/ecdsa" + "errors" + "math/big" + + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/types" +) + +type Account struct { + w *Wallet +} + +func (self *Account) Transact(to *Account, value, gas, price *big.Int, data []byte) error { + return self.w.transact(self, to, value, gas, price, data) +} + +func (self *Account) Address() []byte { + return nil +} + +func (self *Account) PrivateKey() *ecdsa.PrivateKey { + return nil +} + +type Wallet struct{} + +func NewWallet() *Wallet { + return &Wallet{} +} + +func (self *Wallet) GetAccount(i int) *Account { +} + +func (self *Wallet) transact(from, to *Account, value, gas, price *big.Int, data []byte) error { + if from.PrivateKey() == nil { + return errors.New("accounts is not owned (no private key available)") + } + + var createsContract bool + if to == nil { + createsContract = true + } + + var msg *types.Transaction + if contractCreation { + msg = types.NewContractCreationTx(value, gas, price, data) + } else { + msg = types.NewTransactionMessage(to.Address(), value, gas, price, data) + } + + state := self.chainManager.TransState() + nonce := state.GetNonce(key.Address()) + + msg.SetNonce(nonce) + msg.SignECDSA(from.PriateKey()) + + // Do some pre processing for our "pre" events and hooks + block := self.chainManager.NewBlock(from.Address()) + coinbase := state.GetOrNewStateObject(from.Address()) + coinbase.SetGasPool(block.GasLimit()) + self.blockManager.ApplyTransactions(coinbase, state, block, types.Transactions{tx}, true) + + err := self.obj.TxPool().Add(tx) + if err != nil { + return nil, err + } + state.SetNonce(key.Address(), nonce+1) + + if contractCreation { + addr := core.AddressFromMessage(tx) + pipelogger.Infof("Contract addr %x\n", addr) + } + + return tx, nil +} +*/ |