From b91b581b80ec99dfa07b7206104faa919210fc4f Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Thu, 18 Jun 2015 18:00:19 +0300 Subject: eth, eth/fetcher: propagate after header verify, announce only on insert --- eth/backend.go | 2 +- eth/fetcher/fetcher.go | 43 +++++++++++++++++++++----------- eth/fetcher/fetcher_test.go | 18 +++++++------ eth/handler.go | 61 ++++++++++++++++++++++++++++++--------------- 4 files changed, 81 insertions(+), 43 deletions(-) diff --git a/eth/backend.go b/eth/backend.go index c621fa260..37fe66abf 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -313,7 +313,7 @@ func New(config *Config) (*Ethereum, error) { eth.blockProcessor = core.NewBlockProcessor(stateDb, extraDb, eth.pow, eth.chainManager, eth.EventMux()) eth.chainManager.SetProcessor(eth.blockProcessor) - eth.protocolManager = NewProtocolManager(config.ProtocolVersion, config.NetworkId, eth.eventMux, eth.txPool, eth.chainManager) + eth.protocolManager = NewProtocolManager(config.ProtocolVersion, config.NetworkId, eth.eventMux, eth.txPool, eth.pow, eth.chainManager) eth.miner = miner.New(eth, eth.EventMux(), eth.pow) eth.miner.SetGasPrice(config.GasPrice) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index c96471554..d5ff5d77e 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -23,14 +23,17 @@ var ( errTerminated = errors.New("terminated") ) -// hashCheckFn is a callback type for verifying a hash's presence in the local chain. -type hashCheckFn func(common.Hash) bool +// blockRetrievalFn is a callback type for retrieving a block from the local chain. +type blockRetrievalFn func(common.Hash) *types.Block // blockRequesterFn is a callback type for sending a block retrieval request. type blockRequesterFn func([]common.Hash) error +// blockValidatorFn is a callback type to verify a block's header for fast propagation. +type blockValidatorFn func(block *types.Block, parent *types.Block) error + // blockBroadcasterFn is a callback type for broadcasting a block to connected peers. -type blockBroadcasterFn func(block *types.Block) +type blockBroadcasterFn func(block *types.Block, propagate bool) // chainHeightFn is a callback type to retrieve the current chain height. type chainHeightFn func() uint64 @@ -76,7 +79,8 @@ type Fetcher struct { queued map[common.Hash]struct{} // Presence set of already queued blocks (to dedup imports) // Callbacks - hasBlock hashCheckFn // Checks if a block is present in the chain + getBlock blockRetrievalFn // Retrieves a block from the local chain + validateBlock blockValidatorFn // Checks if a block's headers have a valid proof of work broadcastBlock blockBroadcasterFn // Broadcasts a block to connected peers chainHeight chainHeightFn // Retrieves the current chain's height insertChain chainInsertFn // Injects a batch of blocks into the chain @@ -84,7 +88,7 @@ type Fetcher struct { } // New creates a block fetcher to retrieve blocks based on hash announcements. -func New(hasBlock hashCheckFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher { +func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher { return &Fetcher{ notify: make(chan *announce), inject: make(chan *inject), @@ -95,7 +99,8 @@ func New(hasBlock hashCheckFn, broadcastBlock blockBroadcasterFn, chainHeight ch fetching: make(map[common.Hash]*announce), queue: prque.New(), queued: make(map[common.Hash]struct{}), - hasBlock: hasBlock, + getBlock: getBlock, + validateBlock: validateBlock, broadcastBlock: broadcastBlock, chainHeight: chainHeight, insertChain: insertChain, @@ -197,7 +202,7 @@ func (f *Fetcher) loop() { break } // Otherwise if fresh and still unknown, try and import - if number <= height || f.hasBlock(op.block.Hash()) { + if number <= height || f.getBlock(op.block.Hash()) != nil { continue } f.insert(op.origin, op.block) @@ -235,7 +240,7 @@ func (f *Fetcher) loop() { for hash, announces := range f.announced { if time.Since(announces[0].time) > arriveTimeout { announce := announces[rand.Intn(len(announces))] - if !f.hasBlock(hash) { + if f.getBlock(hash) == nil { request[announce.origin] = append(request[announce.origin], hash) f.fetching[hash] = announce } @@ -265,7 +270,7 @@ func (f *Fetcher) loop() { // Filter explicitly requested blocks from hash announcements if _, ok := f.fetching[hash]; ok { // Discard if already imported by other means - if !f.hasBlock(hash) { + if f.getBlock(hash) == nil { explicit = append(explicit, block) } else { delete(f.fetching, hash) @@ -313,7 +318,7 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) { // Discard any past or too distant blocks if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist <= 0 || dist > maxQueueDist { - glog.Infof("Peer %s: discarded block #%d [%x], distance %d", peer, block.NumberU64(), hash.Bytes()[:4], dist) + glog.V(logger.Detail).Infof("Peer %s: discarded block #%d [%x], distance %d", peer, block.NumberU64(), hash.Bytes()[:4], dist) return } // Schedule the block for future importing @@ -321,7 +326,7 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) { f.queued[hash] = struct{}{} f.queue.Push(&inject{origin: peer, block: block}, -float32(block.NumberU64())) - if glog.V(logger.Detail) { + if glog.V(logger.Debug) { glog.Infof("Peer %s: queued block #%d [%x], total %v", peer, block.NumberU64(), hash.Bytes()[:4], f.queue.Size()) } } @@ -339,16 +344,24 @@ func (f *Fetcher) insert(peer string, block *types.Block) { defer func() { f.done <- hash }() // If the parent's unknown, abort insertion - if !f.hasBlock(block.ParentHash()) { + parent := f.getBlock(block.ParentHash()) + if parent == nil { + return + } + // Quickly validate the header and propagate the block if it passes + if err := f.validateBlock(block, parent); err != nil { + glog.V(logger.Debug).Infof("Peer %s: block #%d [%x] verification failed: %v", peer, block.NumberU64(), hash[:4], err) + f.dropPeer(peer) return } + go f.broadcastBlock(block, true) + // Run the actual import and log any issues if _, err := f.insertChain(types.Blocks{block}); err != nil { - glog.V(logger.Detail).Infof("Peer %s: block #%d [%x] import failed: %v", peer, block.NumberU64(), hash[:4], err) - f.dropPeer(peer) + glog.V(logger.Warn).Infof("Peer %s: block #%d [%x] import failed: %v", peer, block.NumberU64(), hash[:4], err) return } // If import succeeded, broadcast the block - go f.broadcastBlock(block) + go f.broadcastBlock(block, false) }() } diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index 3e8df1804..cdf875c5c 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -80,23 +80,27 @@ func newTester() *fetcherTester { hashes: []common.Hash{knownHash}, blocks: map[common.Hash]*types.Block{knownHash: genesis}, } - tester.fetcher = New(tester.hasBlock, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer) + tester.fetcher = New(tester.getBlock, tester.verifyBlock, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer) tester.fetcher.Start() return tester } -// hasBlock checks if a block is pres ent in the testers canonical chain. -func (f *fetcherTester) hasBlock(hash common.Hash) bool { +// getBlock retrieves a block from the tester's block chain. +func (f *fetcherTester) getBlock(hash common.Hash) *types.Block { f.lock.RLock() defer f.lock.RUnlock() - _, ok := f.blocks[hash] - return ok + return f.blocks[hash] +} + +// verifyBlock is a nop placeholder for the block header verification. +func (f *fetcherTester) verifyBlock(block *types.Block, parent *types.Block) error { + return nil } // broadcastBlock is a nop placeholder for the block broadcasting. -func (f *fetcherTester) broadcastBlock(block *types.Block) { +func (f *fetcherTester) broadcastBlock(block *types.Block, propagate bool) { } // chainHeight retrieves the current height (block number) of the chain. @@ -257,7 +261,7 @@ func TestPendingDeduplication(t *testing.T) { return nil } // Announce the same block many times until it's fetched (wait for any pending ops) - for !tester.hasBlock(hashes[0]) { + for tester.getBlock(hashes[0]) == nil { tester.fetcher.Notify("repeater", hashes[0], time.Now().Add(-arriveTimeout), wrapper) time.Sleep(time.Millisecond) } diff --git a/eth/handler.go b/eth/handler.go index f3fe5dfe3..88e340256 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -6,6 +6,8 @@ import ( "sync" "time" + "github.com/ethereum/go-ethereum/pow" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" @@ -67,7 +69,7 @@ type ProtocolManager struct { // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable // with the ethereum network. -func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpool txPool, chainman *core.ChainManager) *ProtocolManager { +func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpool txPool, pow pow.PoW, chainman *core.ChainManager) *ProtocolManager { // Create the protocol manager and initialize peer handlers manager := &ProtocolManager{ eventMux: mux, @@ -91,10 +93,13 @@ func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpo // Construct the different synchronisation mechanisms manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer) + validator := func(block *types.Block, parent *types.Block) error { + return core.ValidateHeader(pow, block.Header(), parent.Header(), true) + } heighter := func() uint64 { return manager.chainman.CurrentBlock().NumberU64() } - manager.fetcher = fetcher.New(manager.chainman.HasBlock, manager.BroadcastBlock, heighter, manager.chainman.InsertChain, manager.removePeer) + manager.fetcher = fetcher.New(manager.chainman.GetBlock, validator, manager.BroadcastBlock, heighter, manager.chainman.InsertChain, manager.removePeer) return manager } @@ -261,6 +266,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { var ( hash common.Hash bytes common.StorageSize + hashes []common.Hash blocks []*types.Block ) for { @@ -270,6 +276,8 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } else if err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } + hashes = append(hashes, hash) + // Retrieve the requested block, stopping if enough was found if block := pm.chainman.GetBlock(hash); block != nil { blocks = append(blocks, block) @@ -279,6 +287,15 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } } + if glog.V(logger.Detail) && len(blocks) == 0 && len(hashes) > 0 { + list := "[" + for _, hash := range hashes { + list += fmt.Sprintf("%x, ", hash[:4]) + } + list = list[:len(list)-2] + "]" + + glog.Infof("Peer %s: no blocks found for requested hashes %s", p.id, list) + } return p.sendBlocks(blocks) case BlocksMsg: @@ -290,6 +307,10 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { glog.V(logger.Detail).Infoln("Decode error", err) blocks = nil } + // Update the receive timestamp of each block + for i:=0; i 0 { pm.downloader.DeliverBlocks(p.id, blocks) @@ -355,28 +376,27 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return nil } -// BroadcastBlock will propagate the block to a subset of its connected peers, -// only notifying the rest of the block's appearance. -func (pm *ProtocolManager) BroadcastBlock(block *types.Block) { +// BroadcastBlock will either propagate a block to a subset of it's peers, or +// will only announce it's availability (depending what's requested). +func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) { hash := block.Hash() - - // Retrieve all the target peers and split between full broadcast or only notification peers := pm.peers.PeersWithoutBlock(hash) - split := int(math.Sqrt(float64(len(peers)))) - - transfer := peers[:split] - notify := peers[split:] - // Send out the data transfers and the notifications - for _, peer := range notify { - peer.sendNewBlockHashes([]common.Hash{hash}) + // If propagation is requested, send to a subset of the peer + if propagate { + transfer := peers[:int(math.Sqrt(float64(len(peers))))] + for _, peer := range transfer { + peer.sendNewBlock(block) + } + glog.V(logger.Detail).Infof("propagated block %x to %d peers in %v", hash[:4], len(transfer), time.Since(block.ReceivedAt)) } - glog.V(logger.Detail).Infof("broadcast hash %x to %d peers.", hash[:4], len(notify)) - - for _, peer := range transfer { - peer.sendNewBlock(block) + // Otherwise if the block is indeed in out own chain, announce it + if pm.chainman.HasBlock(hash) { + for _, peer := range peers { + peer.sendNewBlockHashes([]common.Hash{hash}) + } + glog.V(logger.Detail).Infof("announced block %x to %d peers in %v", hash[:4], len(peers), time.Since(block.ReceivedAt)) } - glog.V(logger.Detail).Infof("broadcast block %x to %d peers. Total processing time: %v", hash[:4], len(transfer), time.Since(block.ReceivedAt)) } // BroadcastTx will propagate the block to its connected peers. It will sort @@ -398,7 +418,8 @@ func (self *ProtocolManager) minedBroadcastLoop() { for obj := range self.minedBlockSub.Chan() { switch ev := obj.(type) { case core.NewMinedBlockEvent: - self.BroadcastBlock(ev.Block) + self.BroadcastBlock(ev.Block, false) + self.BroadcastBlock(ev.Block, true) } } } -- cgit