From ba43364f36db690528cc62196969414ef5e98833 Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 11:41:57 +0200 Subject: [PATCH 1/8] tmp --- block_pool.go | 101 +++++++++++++++++++++++++++++++++++++++++++++++--- ethereum.go | 1 - peer.go | 95 +++++++++++++++++++++++++++++------------------ 3 files changed, 153 insertions(+), 44 deletions(-) diff --git a/block_pool.go b/block_pool.go index 6753308b6..f3a863429 100644 --- a/block_pool.go +++ b/block_pool.go @@ -1,17 +1,23 @@ package eth import ( + "bytes" + "container/list" + "fmt" "math" "math/big" "sync" + "time" "github.com/ethereum/eth-go/ethchain" "github.com/ethereum/eth-go/ethutil" ) type block struct { - peer *Peer - block *ethchain.Block + peer *Peer + block *ethchain.Block + reqAt time.Time + requested int } type BlockPool struct { @@ -22,7 +28,8 @@ type BlockPool struct { hashPool [][]byte pool map[string]*block - td *big.Int + td *big.Int + quit chan bool } func NewBlockPool(eth *Ethereum) *BlockPool { @@ -30,6 +37,7 @@ func NewBlockPool(eth *Ethereum) *BlockPool { eth: eth, pool: make(map[string]*block), td: ethutil.Big0, + quit: make(chan bool), } } @@ -47,7 +55,7 @@ func (self *BlockPool) HasCommonHash(hash []byte) bool { func (self *BlockPool) AddHash(hash []byte) { if self.pool[string(hash)] == nil { - self.pool[string(hash)] = &block{nil, nil} + self.pool[string(hash)] = &block{nil, nil, time.Now(), 0} self.hashPool = append([][]byte{hash}, self.hashPool...) } @@ -58,12 +66,34 @@ func (self *BlockPool) SetBlock(b *ethchain.Block, peer *Peer) { if self.pool[hash] == nil && !self.eth.BlockChain().HasBlock(b.Hash()) { self.hashPool = append(self.hashPool, b.Hash()) - self.pool[hash] = &block{peer, b} + self.pool[hash] = &block{peer, b, time.Now(), 0} } else if self.pool[hash] != nil { self.pool[hash].block = b } } +func (self *BlockPool) getParent(block *ethchain.Block) *ethchain.Block { + for _, item := range self.pool { + if item.block != nil { + if bytes.Compare(item.block.Hash(), block.PrevHash) == 0 { + return item.block + } + } + } + + return nil +} + +func (self *BlockPool) GetChainFromBlock(block *ethchain.Block) ethchain.Blocks { + var blocks ethchain.Blocks + + for b := block; b != nil; b = self.getParent(b) { + blocks = append(ethchain.Blocks{b}, blocks...) + } + + return blocks +} + func (self *BlockPool) CheckLinkAndProcess(f func(block *ethchain.Block)) { var blocks ethchain.Blocks @@ -94,8 +124,14 @@ func (self *BlockPool) Take(amount int, peer *Peer) (hashes [][]byte) { j := 0 for i := 0; i < len(self.hashPool) && j < num; i++ { hash := string(self.hashPool[i]) - if self.pool[hash] != nil && (self.pool[hash].peer == nil || self.pool[hash].peer == peer) && self.pool[hash].block == nil { + item := self.pool[hash] + if item != nil && item.block == nil && + (item.peer == nil || + ((time.Since(item.reqAt) > 5*time.Second && item.peer != peer) && self.eth.peers.Len() > 1) || // multiple peers + (time.Since(item.reqAt) > 5*time.Second && self.eth.peers.Len() == 1) /* single peer*/) { self.pool[hash].peer = peer + self.pool[hash].reqAt = time.Now() + self.pool[hash].requested++ hashes = append(hashes, self.hashPool[i]) j++ @@ -104,3 +140,56 @@ func (self *BlockPool) Take(amount int, peer *Peer) (hashes [][]byte) { return } + +func (self *BlockPool) Start() { + go self.update() +} + +func (self *BlockPool) Stop() { + close(self.quit) +} + +func (self *BlockPool) update() { + serviceTimer := time.NewTicker(100 * time.Millisecond) + procTimer := time.NewTicker(500 * time.Millisecond) +out: + for { + select { + case <-self.quit: + break out + case <-serviceTimer.C: + // Clean up hashes that can't be fetched + done := true + eachPeer(self.eth.peers, func(p *Peer, v *list.Element) { + if p.statusKnown && p.FetchingHashes() { + done = false + } + }) + + if done { + eachPeer(self.eth.peers, func(p *Peer, v *list.Element) { + if p.statusKnown { + hashes := self.Take(100, p) + if len(hashes) > 0 { + p.FetchBlocks(hashes) + if len(hashes) == 1 { + fmt.Printf("last hash = %x\n", hashes[0]) + } else { + fmt.Println("Requesting", len(hashes), "of", p) + } + } + } + }) + } + case <-procTimer.C: + var err error + self.CheckLinkAndProcess(func(block *ethchain.Block) { + err = self.eth.StateManager().Process(block, false) + }) + + if err != nil { + peerlogger.Infoln(err) + } + } + } +} diff --git a/ethereum.go b/ethereum.go index 09665336c..8f667b786 100644 --- a/ethereum.go +++ b/ethereum.go @@ -383,7 +383,6 @@ func (s *Ethereum) ReapDeadPeerHandler() { // Start the ethereum func (s *Ethereum) Start(seed bool) { s.reactor.Start() - s.blockPool.Start() // Bind to addr and port ln, err := net.Listen("tcp", ":"+s.Port) if err != nil { diff --git a/peer.go b/peer.go index 529a35492..8224f4831 100644 --- a/peer.go +++ b/peer.go @@ -131,6 +131,7 @@ type Peer struct { // Last received pong message lastPong int64 lastBlockReceived time.Time + LastHashReceived time.Time host []byte port uint16 @@ -176,6 +177,7 @@ func NewPeer(conn net.Conn, ethereum *Ethereum, inbound bool) *Peer { caps: ethereum.ServerCaps(), version: ethereum.ClientIdentity().String(), protocolCaps: ethutil.NewValue(nil), + td: big.NewInt(0), } } @@ -191,6 +193,7 @@ func NewOutboundPeer(addr string, ethereum *Ethereum, caps Caps) *Peer { caps: caps, version: ethereum.ClientIdentity().String(), protocolCaps: ethutil.NewValue(nil), + td: big.NewInt(0), } // Set up the connection in another goroutine so we don't block the main thread @@ -505,6 +508,9 @@ func (p *Peer) HandleInbound() { for it.Next() { hash := it.Value().Bytes() + p.lastReceivedHash = hash + p.LastHashReceived = time.Now() + if blockPool.HasCommonHash(hash) { foundCommonHash = true @@ -512,15 +518,16 @@ func (p *Peer) HandleInbound() { } blockPool.AddHash(hash) - - p.lastReceivedHash = hash - - p.lastBlockReceived = time.Now() } - if foundCommonHash || msg.Data.Len() == 0 { - p.FetchBlocks() - } else { + /* + if foundCommonHash || msg.Data.Len() == 0 { + p.FetchBlocks() + } else { + p.FetchHashes() + } + */ + if !foundCommonHash && msg.Data.Len() != 0 { p.FetchHashes() } @@ -539,19 +546,21 @@ func (p *Peer) HandleInbound() { p.lastBlockReceived = time.Now() } - var err error - blockPool.CheckLinkAndProcess(func(block *ethchain.Block) { - err = p.ethereum.StateManager().Process(block, false) - }) + /* + var err error + blockPool.CheckLinkAndProcess(func(block *ethchain.Block) { + err = p.ethereum.StateManager().Process(block, false) + }) - if err != nil { - peerlogger.Infoln(err) - } else { - // Don't trigger if there's just one block. - if blockPool.Len() != 0 && msg.Data.Len() > 1 { - p.FetchBlocks() + if err != nil { + peerlogger.Infoln(err) + } else { + // Don't trigger if there's just one block. + if blockPool.Len() != 0 && msg.Data.Len() > 1 { + p.FetchBlocks() + } } - } + */ } } } @@ -560,10 +569,7 @@ func (p *Peer) HandleInbound() { p.Stop() } -func (self *Peer) FetchBlocks() { - blockPool := self.ethereum.blockPool - - hashes := blockPool.Take(100, self) +func (self *Peer) FetchBlocks(hashes [][]byte) { if len(hashes) > 0 { self.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlocksTy, ethutil.ByteSliceToInterface(hashes))) } @@ -572,7 +578,7 @@ func (self *Peer) FetchBlocks() { func (self *Peer) FetchHashes() { blockPool := self.ethereum.blockPool - if self.td.Cmp(blockPool.td) >= 0 { + if self.td.Cmp(self.ethereum.HighestTDPeer()) >= 0 { blockPool.td = self.td if !blockPool.HasLatestHash() { @@ -581,6 +587,10 @@ func (self *Peer) FetchHashes() { } } +func (self *Peer) FetchingHashes() bool { + return time.Since(self.LastHashReceived) < 5*time.Second +} + // General update method func (self *Peer) update() { serviceTimer := time.NewTicker(5 * time.Second) @@ -589,11 +599,22 @@ out: for { select { case <-serviceTimer.C: - since := time.Since(self.lastBlockReceived) - if since > 10*time.Second && self.ethereum.blockPool.Len() != 0 && self.IsCap("eth") { - self.FetchHashes() - } else if since > 5*time.Second { - self.catchingUp = false + if self.IsCap("eth") { + var ( + sinceBlock = time.Since(self.lastBlockReceived) + sinceHash = time.Since(self.LastHashReceived) + ) + + if sinceBlock > 5*time.Second && sinceHash > 5*time.Second { + self.catchingUp = false + } + + if sinceHash > 10*time.Second && self.ethereum.blockPool.Len() != 0 { + // XXX While this is completely and utterly incorrect, in order to do anything on the test net is to do it this way + // Assume that when fetching hashes timeouts, we are done. + //self.FetchHashes() + //self.FetchBlocks() + } } case <-self.quit: break out @@ -761,6 +782,14 @@ func (p *Peer) handleHandshake(msg *ethwire.Msg) { return } + // Self connect detection + pubkey := p.ethereum.KeyManager().PublicKey() + if bytes.Compare(pubkey[1:], pub) == 0 { + p.Stop() + + return + } + usedPub := 0 // This peer is already added to the peerlist so we expect to find a double pubkey at least once eachPeer(p.ethereum.Peers(), func(peer *Peer, e *list.Element) { @@ -779,16 +808,8 @@ func (p *Peer) handleHandshake(msg *ethwire.Msg) { // If this is an inbound connection send an ack back if p.inbound { p.port = uint16(port) - - // Self connect detection - pubkey := p.ethereum.KeyManager().PublicKey() - if bytes.Compare(pubkey, p.pubkey) == 0 { - p.Stop() - - return - } - } + p.SetVersion(clientId) p.versionKnown = true From ad13b402d715e120009b3efb6a08d2c90139f31d Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 17:47:01 +0200 Subject: [PATCH 2/8] Fixed race condition --- ethtrie/trie.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ethtrie/trie.go b/ethtrie/trie.go index e6e09dd0d..695ff436a 100644 --- a/ethtrie/trie.go +++ b/ethtrie/trie.go @@ -196,8 +196,8 @@ func (t *Trie) Update(key, value string) { } func (t *Trie) Get(key string) string { - t.mut.RLock() - defer t.mut.RUnlock() + t.mut.Lock() + defer t.mut.Unlock() k := CompactHexDecode(key) c := ethutil.NewValue(t.getState(t.Root, k)) From d3a0bb4f3535dda89c3cb1d26ff3168faaaf7ca6 Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 19:54:14 +0200 Subject: [PATCH 3/8] Info => Debug --- ethchain/state_manager.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/ethchain/state_manager.go b/ethchain/state_manager.go index b0ea754f4..888c68bb9 100644 --- a/ethchain/state_manager.go +++ b/ethchain/state_manager.go @@ -250,15 +250,13 @@ func (sm *StateManager) Process(block *Block, dontReact bool) (err error) { fk := append([]byte("bloom"), block.Hash()...) sm.Ethereum.Db().Put(fk, filter.Bin()) - statelogger.Infof("Added block #%d (%x)\n", block.Number, block.Hash()) + statelogger.Debugf("Added block #%d (%x)\n", block.Number, block.Hash()) if dontReact == false { sm.Ethereum.Reactor().Post("newBlock", block) state.Manifest().Reset() } - sm.Ethereum.Broadcast(ethwire.MsgBlockTy, []interface{}{block.Value().Val}) - sm.Ethereum.TxPool().RemoveInvalid(state) } else { statelogger.Errorln("total diff failed") From 84690bfbbe99b0c8b8d3a377d39b0900990b300c Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 19:54:30 +0200 Subject: [PATCH 4/8] Changed the block fetching code and hash distribution --- block_pool.go | 128 +++++++++++++++++++++++++++++++++----------------- 1 file changed, 84 insertions(+), 44 deletions(-) diff --git a/block_pool.go b/block_pool.go index f3a863429..88d1c3739 100644 --- a/block_pool.go +++ b/block_pool.go @@ -3,17 +3,21 @@ package eth import ( "bytes" "container/list" - "fmt" "math" "math/big" "sync" "time" "github.com/ethereum/eth-go/ethchain" + "github.com/ethereum/eth-go/ethlog" "github.com/ethereum/eth-go/ethutil" + "github.com/ethereum/eth-go/ethwire" ) +var poollogger = ethlog.NewLogger("[BPOOL]") + type block struct { + from *Peer peer *Peer block *ethchain.Block reqAt time.Time @@ -30,6 +34,8 @@ type BlockPool struct { td *big.Int quit chan bool + + ChainLength, BlocksProcessed int } func NewBlockPool(eth *Ethereum) *BlockPool { @@ -53,9 +59,9 @@ func (self *BlockPool) HasCommonHash(hash []byte) bool { return self.eth.BlockChain().GetBlock(hash) != nil } -func (self *BlockPool) AddHash(hash []byte) { +func (self *BlockPool) AddHash(hash []byte, peer *Peer) { if self.pool[string(hash)] == nil { - self.pool[string(hash)] = &block{nil, nil, time.Now(), 0} + self.pool[string(hash)] = &block{peer, nil, nil, time.Now(), 0} self.hashPool = append([][]byte{hash}, self.hashPool...) } @@ -66,10 +72,12 @@ func (self *BlockPool) SetBlock(b *ethchain.Block, peer *Peer) { if self.pool[hash] == nil && !self.eth.BlockChain().HasBlock(b.Hash()) { self.hashPool = append(self.hashPool, b.Hash()) - self.pool[hash] = &block{peer, b, time.Now(), 0} + self.pool[hash] = &block{peer, peer, b, time.Now(), 0} } else if self.pool[hash] != nil { self.pool[hash].block = b } + + self.BlocksProcessed++ } func (self *BlockPool) getParent(block *ethchain.Block) *ethchain.Block { @@ -94,18 +102,24 @@ func (self *BlockPool) GetChainFromBlock(block *ethchain.Block) ethchain.Blocks return blocks } -func (self *BlockPool) CheckLinkAndProcess(f func(block *ethchain.Block)) { - - var blocks ethchain.Blocks +func (self *BlockPool) Blocks() (blocks ethchain.Blocks) { for _, item := range self.pool { if item.block != nil { blocks = append(blocks, item.block) } } + return +} + +func (self *BlockPool) ProcessCanonical(f func(block *ethchain.Block)) (procAmount int) { + blocks := self.Blocks() + ethchain.BlockBy(ethchain.Number).Sort(blocks) for _, block := range blocks { if self.eth.BlockChain().HasBlock(block.PrevHash) { + procAmount++ + f(block) hash := block.Hash() @@ -114,31 +128,58 @@ func (self *BlockPool) CheckLinkAndProcess(f func(block *ethchain.Block)) { } } + + return } -func (self *BlockPool) Take(amount int, peer *Peer) (hashes [][]byte) { - self.mut.Lock() - defer self.mut.Unlock() +func (self *BlockPool) DistributeHashes() { + var ( + peerLen = self.eth.peers.Len() + amount = 200 * peerLen + dist = make(map[*Peer][][]byte) + ) num := int(math.Min(float64(amount), float64(len(self.pool)))) - j := 0 - for i := 0; i < len(self.hashPool) && j < num; i++ { - hash := string(self.hashPool[i]) - item := self.pool[hash] - if item != nil && item.block == nil && - (item.peer == nil || - ((time.Since(item.reqAt) > 5*time.Second && item.peer != peer) && self.eth.peers.Len() > 1) || // multiple peers - (time.Since(item.reqAt) > 5*time.Second && self.eth.peers.Len() == 1) /* single peer*/) { - self.pool[hash].peer = peer - self.pool[hash].reqAt = time.Now() - self.pool[hash].requested++ + for i, j := 0, 0; i < len(self.hashPool) && j < num; i++ { + hash := self.hashPool[i] + item := self.pool[string(hash)] - hashes = append(hashes, self.hashPool[i]) - j++ + if item != nil && item.block == nil { + var peer *Peer + lastFetchFailed := time.Since(item.reqAt) > 5*time.Second + + // Handle failed requests + if lastFetchFailed && item.requested > 0 && item.peer != nil { + if item.requested < 100 { + // Select peer the hash was retrieved off + peer = item.from + } else { + // Remove it + self.hashPool = ethutil.DeleteFromByteSlice(self.hashPool, hash) + delete(self.pool, string(hash)) + } + } else if lastFetchFailed || item.peer == nil { + // Find a suitable, available peer + eachPeer(self.eth.peers, func(p *Peer, v *list.Element) { + if peer == nil && len(dist[p]) < amount/peerLen { + peer = p + } + }) + } + + if peer != nil { + item.reqAt = time.Now() + item.peer = peer + item.requested++ + + dist[peer] = append(dist[peer], hash) + } } } - return + for peer, hashes := range dist { + peer.FetchBlocks(hashes) + } } func (self *BlockPool) Start() { @@ -158,7 +199,8 @@ out: case <-self.quit: break out case <-serviceTimer.C: - // Clean up hashes that can't be fetched + // Check if we're catching up. If not distribute the hashes to + // the peers and download the blockchain done := true eachPeer(self.eth.peers, func(p *Peer, v *list.Element) { if p.statusKnown && p.FetchingHashes() { @@ -166,29 +208,27 @@ out: } }) - if done { - eachPeer(self.eth.peers, func(p *Peer, v *list.Element) { - if p.statusKnown { - hashes := self.Take(100, p) - if len(hashes) > 0 { - p.FetchBlocks(hashes) - if len(hashes) == 1 { - fmt.Printf("last hash = %x\n", hashes[0]) - } else { - fmt.Println("Requesting", len(hashes), "of", p) - } - } - } - }) + if done && len(self.hashPool) > 0 { + self.DistributeHashes() + } + + if self.ChainLength < len(self.hashPool) { + self.ChainLength = len(self.hashPool) } case <-procTimer.C: - var err error - self.CheckLinkAndProcess(func(block *ethchain.Block) { - err = self.eth.StateManager().Process(block, false) + // XXX We can optimize this lifting this on to a new goroutine. + // We'd need to make sure that the pools are properly protected by a mutex + amount := self.ProcessCanonical(func(block *ethchain.Block) { + err := self.eth.StateManager().Process(block, false) + if err != nil { + poollogger.Infoln(err) + } }) - if err != nil { - peerlogger.Infoln(err) + // Do not propagate to the network on catchups + if amount == 1 { + block := self.eth.BlockChain().CurrentBlock + self.eth.Broadcast(ethwire.MsgBlockTy, []interface{}{block.Value().Val}) } } } From 60a8c9527cba4298be4bdde39ff1e0e3fd7fe637 Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 19:54:37 +0200 Subject: [PATCH 5/8] Mine without txs --- ethminer/miner.go | 7 ------- 1 file changed, 7 deletions(-) diff --git a/ethminer/miner.go b/ethminer/miner.go index b74f90e0c..299a5204a 100644 --- a/ethminer/miner.go +++ b/ethminer/miner.go @@ -3,7 +3,6 @@ package ethminer import ( "bytes" "sort" - "time" "github.com/ethereum/eth-go/ethchain" "github.com/ethereum/eth-go/ethlog" @@ -136,12 +135,6 @@ func (miner *Miner) listener() { } } default: - // This hack is only temporarily - if len(miner.txs) == 0 { - time.Sleep(2 * time.Second) - continue - } - miner.mineNewBlock() } } From 54558dd8ae55fc2c7e226c908222803671b7feeb Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 19:54:51 +0200 Subject: [PATCH 6/8] Removed log --- ethvm/vm.go | 1 - 1 file changed, 1 deletion(-) diff --git a/ethvm/vm.go b/ethvm/vm.go index 7aff320f9..bb14d75da 100644 --- a/ethvm/vm.go +++ b/ethvm/vm.go @@ -961,7 +961,6 @@ func (self *Message) Addr() []byte { } func (self *Message) Exec(codeAddr []byte, caller ClosureRef) (ret []byte, err error) { - fmt.Printf("%x %x\n", codeAddr[0:4], self.address[0:4]) queue := self.vm.queue self.vm.queue = list.New() From 1fe2d0d0e011b6d6bc6261b69fe955943d9d6e3e Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 19:55:28 +0200 Subject: [PATCH 7/8] Peers no longer take care of block processing --- peer.go | 34 ++-------------------------------- 1 file changed, 2 insertions(+), 32 deletions(-) diff --git a/peer.go b/peer.go index 8224f4831..3bbfb74fa 100644 --- a/peer.go +++ b/peer.go @@ -517,16 +517,9 @@ func (p *Peer) HandleInbound() { break } - blockPool.AddHash(hash) + blockPool.AddHash(hash, p) } - /* - if foundCommonHash || msg.Data.Len() == 0 { - p.FetchBlocks() - } else { - p.FetchHashes() - } - */ if !foundCommonHash && msg.Data.Len() != 0 { p.FetchHashes() } @@ -545,22 +538,6 @@ func (p *Peer) HandleInbound() { p.lastBlockReceived = time.Now() } - - /* - var err error - blockPool.CheckLinkAndProcess(func(block *ethchain.Block) { - err = p.ethereum.StateManager().Process(block, false) - }) - - if err != nil { - peerlogger.Infoln(err) - } else { - // Don't trigger if there's just one block. - if blockPool.Len() != 0 && msg.Data.Len() > 1 { - p.FetchBlocks() - } - } - */ } } } @@ -608,13 +585,6 @@ out: if sinceBlock > 5*time.Second && sinceHash > 5*time.Second { self.catchingUp = false } - - if sinceHash > 10*time.Second && self.ethereum.blockPool.Len() != 0 { - // XXX While this is completely and utterly incorrect, in order to do anything on the test net is to do it this way - // Assume that when fetching hashes timeouts, we are done. - //self.FetchHashes() - //self.FetchBlocks() - } } case <-self.quit: break out @@ -738,7 +708,7 @@ func (self *Peer) handleStatus(msg *ethwire.Msg) { // Compare the total TD with the blockchain TD. If remote is higher // fetch hashes from highest TD node. if self.td.Cmp(self.ethereum.BlockChain().TD) > 0 { - self.ethereum.blockPool.AddHash(self.lastReceivedHash) + self.ethereum.blockPool.AddHash(self.lastReceivedHash, self) self.FetchHashes() } From cba2f6c2c472d5818541f6eabf3e18196d0c477d Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 19:56:21 +0200 Subject: [PATCH 8/8] Block pool stop / start --- ethereum.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/ethereum.go b/ethereum.go index 8f667b786..d04b08960 100644 --- a/ethereum.go +++ b/ethereum.go @@ -158,6 +158,9 @@ func (s *Ethereum) StateManager() *ethchain.StateManager { func (s *Ethereum) TxPool() *ethchain.TxPool { return s.txPool } +func (s *Ethereum) BlockPool() *BlockPool { + return s.blockPool +} func (self *Ethereum) Db() ethutil.Database { return self.db } @@ -383,6 +386,7 @@ func (s *Ethereum) ReapDeadPeerHandler() { // Start the ethereum func (s *Ethereum) Start(seed bool) { s.reactor.Start() + s.blockPool.Start() // Bind to addr and port ln, err := net.Listen("tcp", ":"+s.Port) if err != nil { @@ -503,6 +507,7 @@ func (s *Ethereum) Stop() { s.stateManager.Stop() s.reactor.Flush() s.reactor.Stop() + s.blockPool.Stop() ethlogger.Infoln("Server stopped") close(s.shutdownChan)