From 7c2af1c11722dc3175a98342c060afcfaf6a275f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 16 Jun 2015 11:58:32 +0300 Subject: eth, eth/fetcher: separate notification sync mechanism --- eth/handler.go | 71 +++++++++++++++++++++------------------------------------- 1 file changed, 26 insertions(+), 45 deletions(-) (limited to 'eth/handler.go') diff --git a/eth/handler.go b/eth/handler.go index ec4f2d53a..99ac4ce68 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -7,6 +7,8 @@ import ( "sync" "time" + "github.com/ethereum/go-ethereum/eth/fetcher" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" @@ -45,6 +47,7 @@ type ProtocolManager struct { txpool txPool chainman *core.ChainManager downloader *downloader.Downloader + fetcher *fetcher.Fetcher peers *peerSet SubProtocol p2p.Protocol @@ -54,11 +57,9 @@ type ProtocolManager struct { minedBlockSub event.Subscription // channels for fetcher, syncer, txsyncLoop - newPeerCh chan *peer - newHashCh chan []*blockAnnounce - newBlockCh chan chan []*types.Block - txsyncCh chan *txsync - quitSync chan struct{} + newPeerCh chan *peer + txsyncCh chan *txsync + quitSync chan struct{} // wait group is used for graceful shutdowns during downloading // and processing @@ -69,30 +70,33 @@ 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 { + // Create the protocol manager and initialize peer handlers manager := &ProtocolManager{ - eventMux: mux, - txpool: txpool, - chainman: chainman, - peers: newPeerSet(), - newPeerCh: make(chan *peer, 1), - newHashCh: make(chan []*blockAnnounce, 1), - newBlockCh: make(chan chan []*types.Block), - txsyncCh: make(chan *txsync), - quitSync: make(chan struct{}), + eventMux: mux, + txpool: txpool, + chainman: chainman, + peers: newPeerSet(), + newPeerCh: make(chan *peer, 1), + txsyncCh: make(chan *txsync), + quitSync: make(chan struct{}), } - manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer) manager.SubProtocol = p2p.Protocol{ Name: "eth", Version: uint(protocolVersion), Length: ProtocolLength, Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { peer := manager.newPeer(protocolVersion, networkId, p, rw) - manager.newPeerCh <- peer - return manager.handle(peer) }, } + // Construct the different synchronisation mechanisms + manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer) + + importer := func(peer string, block *types.Block) error { + return manager.importBlock(manager.peers.Peer(peer), block, nil) + } + manager.fetcher = fetcher.New(manager.chainman.HasBlock, importer) return manager } @@ -126,7 +130,6 @@ func (pm *ProtocolManager) Start() { // start sync handlers go pm.syncer() - go pm.fetcher() go pm.txsyncLoop() } @@ -291,20 +294,9 @@ func (self *ProtocolManager) handleMsg(p *peer) error { glog.V(logger.Detail).Infoln("Decode error", err) blocks = nil } - // Filter out any explicitly requested blocks (cascading select to get blocking back to peer) - filter := make(chan []*types.Block) - select { - case <-self.quitSync: - case self.newBlockCh <- filter: - select { - case <-self.quitSync: - case filter <- blocks: - select { - case <-self.quitSync: - case blocks := <-filter: - self.downloader.DeliverBlocks(p.id, blocks) - } - } + // Filter out any explicitly requested blocks, deliver the rest to the downloader + if blocks := self.fetcher.Filter(blocks); len(blocks) > 0 { + self.downloader.DeliverBlocks(p.id, blocks) } case NewBlockHashesMsg: @@ -327,19 +319,8 @@ func (self *ProtocolManager) handleMsg(p *peer) error { unknown = append(unknown, hash) } } - announces := make([]*blockAnnounce, len(unknown)) - for i, hash := range unknown { - announces[i] = &blockAnnounce{ - hash: hash, - peer: p, - time: time.Now(), - } - } - if len(announces) > 0 { - select { - case self.newHashCh <- announces: - case <-self.quitSync: - } + for _, hash := range unknown { + self.fetcher.Notify(p.id, hash, time.Now(), p.requestBlocks) } case NewBlockMsg: -- cgit v1.2.3 From 057bc237adf9ed0adf615a72cc1e92d9aed15d9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 16 Jun 2015 17:39:04 +0300 Subject: eth, eth/fetcher: use an import queue to store out of order blocks --- eth/handler.go | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) (limited to 'eth/handler.go') diff --git a/eth/handler.go b/eth/handler.go index 99ac4ce68..e5908dc88 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -1,6 +1,7 @@ package eth import ( + "errors" "fmt" "math" "math/big" @@ -94,9 +95,15 @@ func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpo manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer) importer := func(peer string, block *types.Block) error { - return manager.importBlock(manager.peers.Peer(peer), block, nil) + if p := manager.peers.Peer(peer); p != nil { + return manager.importBlock(manager.peers.Peer(peer), block, nil) + } + return errors.New("unknown peer") + } + heighter := func() uint64 { + return manager.chainman.CurrentBlock().NumberU64() } - manager.fetcher = fetcher.New(manager.chainman.HasBlock, importer) + manager.fetcher = fetcher.New(manager.chainman.HasBlock, importer, heighter) return manager } -- cgit v1.2.3 From 11c8f83a583745b01a4c06cdd29529af1df364f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 16 Jun 2015 18:14:52 +0300 Subject: eth, eth/fetcher: cache future propagated blocks too --- eth/handler.go | 2 ++ 1 file changed, 2 insertions(+) (limited to 'eth/handler.go') diff --git a/eth/handler.go b/eth/handler.go index e5908dc88..604983a7b 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -340,6 +340,8 @@ func (self *ProtocolManager) handleMsg(p *peer) error { } request.Block.ReceivedAt = msg.ReceivedAt + // Try to import the propagated block, also making it fill any fetcher gaps + self.fetcher.Enqueue(p.id, request.Block) if err := self.importBlock(p, request.Block, request.TD); err != nil { return err } -- cgit v1.2.3 From 497a7f1717a798e59c9550fbb58504b08fe13b21 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 16 Jun 2015 23:18:01 +0300 Subject: eth, eth/fetcher: define and enforce propagation boundaries --- eth/handler.go | 1 - 1 file changed, 1 deletion(-) (limited to 'eth/handler.go') diff --git a/eth/handler.go b/eth/handler.go index 604983a7b..c3b58650d 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -9,7 +9,6 @@ import ( "time" "github.com/ethereum/go-ethereum/eth/fetcher" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" -- cgit v1.2.3 From 37c5ff392f71fddaa6acd92925f69e81876fe22e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 17 Jun 2015 16:53:28 +0300 Subject: eth/fetcher: build longest chain until proven otherwise --- eth/handler.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'eth/handler.go') diff --git a/eth/handler.go b/eth/handler.go index c3b58650d..b62815532 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -8,11 +8,11 @@ import ( "sync" "time" - "github.com/ethereum/go-ethereum/eth/fetcher" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth/downloader" + "github.com/ethereum/go-ethereum/eth/fetcher" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" -- cgit v1.2.3 From 5ec6ecc511dc861c0395335f06883bc9ac650e4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 17 Jun 2015 18:25:23 +0300 Subject: eth, eth/fetcher: move propagated block import into fetcher --- eth/handler.go | 110 +++++++++++++++++---------------------------------------- 1 file changed, 33 insertions(+), 77 deletions(-) (limited to 'eth/handler.go') diff --git a/eth/handler.go b/eth/handler.go index b62815532..eeb6a4f14 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -1,10 +1,8 @@ package eth import ( - "errors" "fmt" "math" - "math/big" "sync" "time" @@ -93,16 +91,10 @@ 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) - importer := func(peer string, block *types.Block) error { - if p := manager.peers.Peer(peer); p != nil { - return manager.importBlock(manager.peers.Peer(peer), block, nil) - } - return errors.New("unknown peer") - } heighter := func() uint64 { return manager.chainman.CurrentBlock().NumberU64() } - manager.fetcher = fetcher.New(manager.chainman.HasBlock, importer, heighter) + manager.fetcher = fetcher.New(manager.chainman.HasBlock, manager.BroadcastBlock, heighter, manager.chainman.InsertChain, manager.removePeer) return manager } @@ -194,7 +186,7 @@ func (pm *ProtocolManager) handle(p *peer) error { return nil } -func (self *ProtocolManager) handleMsg(p *peer) error { +func (pm *ProtocolManager) handleMsg(p *peer) error { msg, err := p.rw.ReadMsg() if err != nil { return err @@ -224,7 +216,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error { RemoteId: p.ID().String(), }) } - self.txpool.AddTransactions(txs) + pm.txpool.AddTransactions(txs) case GetBlockHashesMsg: var request getBlockHashesMsgData @@ -236,7 +228,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error { request.Amount = uint64(downloader.MaxHashFetch) } - hashes := self.chainman.GetBlockHashesFromHash(request.Hash, request.Amount) + hashes := pm.chainman.GetBlockHashesFromHash(request.Hash, request.Amount) if glog.V(logger.Debug) { if len(hashes) == 0 { @@ -254,7 +246,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error { if err := msgStream.Decode(&hashes); err != nil { break } - err := self.downloader.DeliverHashes(p.id, hashes) + err := pm.downloader.DeliverHashes(p.id, hashes) if err != nil { glog.V(logger.Debug).Infoln(err) } @@ -280,7 +272,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrDecode, "msg %v: %v", msg, err) } - block := self.chainman.GetBlock(hash) + block := pm.chainman.GetBlock(hash) if block != nil { blocks = append(blocks, block) totalsize += block.Size() @@ -301,8 +293,8 @@ func (self *ProtocolManager) handleMsg(p *peer) error { blocks = nil } // Filter out any explicitly requested blocks, deliver the rest to the downloader - if blocks := self.fetcher.Filter(blocks); len(blocks) > 0 { - self.downloader.DeliverBlocks(p.id, blocks) + if blocks := pm.fetcher.Filter(blocks); len(blocks) > 0 { + pm.downloader.DeliverBlocks(p.id, blocks) } case NewBlockHashesMsg: @@ -321,15 +313,16 @@ func (self *ProtocolManager) handleMsg(p *peer) error { // Schedule all the unknown hashes for retrieval unknown := make([]common.Hash, 0, len(hashes)) for _, hash := range hashes { - if !self.chainman.HasBlock(hash) { + if !pm.chainman.HasBlock(hash) { unknown = append(unknown, hash) } } for _, hash := range unknown { - self.fetcher.Notify(p.id, hash, time.Now(), p.requestBlocks) + pm.fetcher.Notify(p.id, hash, time.Now(), p.requestBlocks) } case NewBlockMsg: + // Retrieve and decode the propagated block var request newBlockMsgData if err := msg.Decode(&request); err != nil { return errResp(ErrDecode, "%v: %v", msg, err) @@ -339,11 +332,24 @@ func (self *ProtocolManager) handleMsg(p *peer) error { } request.Block.ReceivedAt = msg.ReceivedAt - // Try to import the propagated block, also making it fill any fetcher gaps - self.fetcher.Enqueue(p.id, request.Block) - if err := self.importBlock(p, request.Block, request.TD); err != nil { - return err - } + // Mark the block's arrival for whatever reason + _, chainHead, _ := pm.chainman.Status() + jsonlogger.LogJson(&logger.EthChainReceivedNewBlock{ + BlockHash: request.Block.Hash().Hex(), + BlockNumber: request.Block.Number(), + ChainHeadHash: chainHead.Hex(), + BlockPrevHash: request.Block.ParentHash().Hex(), + RemoteId: p.ID().String(), + }) + // Mark the peer as owning the block and schedule it for import + p.blockHashes.Add(request.Block.Hash()) + p.SetHead(request.Block.Hash()) + + pm.fetcher.Enqueue(p.id, request.Block) + + // TODO: Schedule a sync to cover potential gaps (this needs proto update) + p.SetTd(request.TD) + go pm.synchronise(p) default: return errResp(ErrInvalidMsgCode, "%v", msg.Code) @@ -351,61 +357,11 @@ func (self *ProtocolManager) handleMsg(p *peer) error { return nil } -// importBlocks injects a new block retrieved from the given peer into the chain -// manager. -func (pm *ProtocolManager) importBlock(p *peer, block *types.Block, td *big.Int) error { - hash := block.Hash() - - // Mark the block as present at the remote node (don't duplicate already held data) - p.blockHashes.Add(hash) - p.SetHead(hash) - if td != nil { - p.SetTd(td) - } - // Log the block's arrival - _, chainHead, _ := pm.chainman.Status() - jsonlogger.LogJson(&logger.EthChainReceivedNewBlock{ - BlockHash: hash.Hex(), - BlockNumber: block.Number(), - ChainHeadHash: chainHead.Hex(), - BlockPrevHash: block.ParentHash().Hex(), - RemoteId: p.ID().String(), - }) - // If the block's already known or its difficulty is lower than ours, drop - if pm.chainman.HasBlock(hash) { - p.SetTd(pm.chainman.GetBlock(hash).Td) // update the peer's TD to the real value - return nil - } - if td != nil && pm.chainman.Td().Cmp(td) > 0 && new(big.Int).Add(block.Number(), big.NewInt(7)).Cmp(pm.chainman.CurrentBlock().Number()) < 0 { - glog.V(logger.Debug).Infof("[%s] dropped block %v due to low TD %v\n", p.id, block.Number(), td) - return nil - } - // Attempt to insert the newly received block and propagate to our peers - if pm.chainman.HasBlock(block.ParentHash()) { - if _, err := pm.chainman.InsertChain(types.Blocks{block}); err != nil { - glog.V(logger.Error).Infoln("removed peer (", p.id, ") due to block error", err) - return err - } - if td != nil && block.Td.Cmp(td) != 0 { - err := fmt.Errorf("invalid TD on block(%v) from peer(%s): block.td=%v, request.td=%v", block.Number(), p.id, block.Td, td) - glog.V(logger.Error).Infoln(err) - return err - } - pm.BroadcastBlock(hash, block) - return nil - } - // Parent of the block is unknown, try to sync with this peer if it seems to be good - if td != nil { - go pm.synchronise(p) - } - 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(hash common.Hash, block *types.Block) { +func (pm *ProtocolManager) BroadcastBlock(block *types.Block) { // Retrieve all the target peers and split between full broadcast or only notification - peers := pm.peers.PeersWithoutBlock(hash) + peers := pm.peers.PeersWithoutBlock(block.Hash()) split := int(math.Sqrt(float64(len(peers)))) transfer := peers[:split] @@ -413,7 +369,7 @@ func (pm *ProtocolManager) BroadcastBlock(hash common.Hash, block *types.Block) // Send out the data transfers and the notifications for _, peer := range notify { - peer.sendNewBlockHashes([]common.Hash{hash}) + peer.sendNewBlockHashes([]common.Hash{block.Hash()}) } glog.V(logger.Detail).Infoln("broadcast hash to", len(notify), "peers.") @@ -442,7 +398,7 @@ func (self *ProtocolManager) minedBroadcastLoop() { for obj := range self.minedBlockSub.Chan() { switch ev := obj.(type) { case core.NewMinedBlockEvent: - self.BroadcastBlock(ev.Block.Hash(), ev.Block) + self.BroadcastBlock(ev.Block) } } } -- cgit v1.2.3 From 629705ad53ceac1bcc5f32f2d7ef51d7626aefef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 18 Jun 2015 16:09:34 +0300 Subject: eth: clean the block request packet handling a bit --- eth/handler.go | 34 +++++++++++++++++----------------- 1 file changed, 17 insertions(+), 17 deletions(-) (limited to 'eth/handler.go') diff --git a/eth/handler.go b/eth/handler.go index eeb6a4f14..f3fe5dfe3 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -252,33 +252,31 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } case GetBlocksMsg: - var blocks []*types.Block - + // Decode the retrieval message msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size)) if _, err := msgStream.List(); err != nil { return err } + // Gather blocks until the fetch or network limits is reached var ( - i int - totalsize common.StorageSize + hash common.Hash + bytes common.StorageSize + blocks []*types.Block ) for { - i++ - var hash common.Hash err := msgStream.Decode(&hash) if err == rlp.EOL { break } else if err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - - block := pm.chainman.GetBlock(hash) - if block != nil { + // Retrieve the requested block, stopping if enough was found + if block := pm.chainman.GetBlock(hash); block != nil { blocks = append(blocks, block) - totalsize += block.Size() - } - if i == downloader.MaxBlockFetch || totalsize > maxBlockRespSize { - break + bytes += block.Size() + if len(blocks) >= downloader.MaxBlockFetch || bytes > maxBlockRespSize { + break + } } } return p.sendBlocks(blocks) @@ -360,8 +358,10 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { // 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) { + hash := block.Hash() + // Retrieve all the target peers and split between full broadcast or only notification - peers := pm.peers.PeersWithoutBlock(block.Hash()) + peers := pm.peers.PeersWithoutBlock(hash) split := int(math.Sqrt(float64(len(peers)))) transfer := peers[:split] @@ -369,14 +369,14 @@ func (pm *ProtocolManager) BroadcastBlock(block *types.Block) { // Send out the data transfers and the notifications for _, peer := range notify { - peer.sendNewBlockHashes([]common.Hash{block.Hash()}) + peer.sendNewBlockHashes([]common.Hash{hash}) } - glog.V(logger.Detail).Infoln("broadcast hash to", len(notify), "peers.") + glog.V(logger.Detail).Infof("broadcast hash %x to %d peers.", hash[:4], len(notify)) for _, peer := range transfer { peer.sendNewBlock(block) } - glog.V(logger.Detail).Infoln("broadcast block to", len(transfer), "peers. Total processing time:", 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 -- cgit v1.2.3 From b91b581b80ec99dfa07b7206104faa919210fc4f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 18 Jun 2015 18:00:19 +0300 Subject: eth, eth/fetcher: propagate after header verify, announce only on insert --- eth/handler.go | 61 +++++++++++++++++++++++++++++++++++++++------------------- 1 file changed, 41 insertions(+), 20 deletions(-) (limited to 'eth/handler.go') 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 v1.2.3 From 90d45f0397666cce42657849257a6f4f7cd2381c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 18 Jun 2015 18:25:27 +0300 Subject: eth: fix test breakage from the previous commit --- eth/handler.go | 1 - 1 file changed, 1 deletion(-) (limited to 'eth/handler.go') diff --git a/eth/handler.go b/eth/handler.go index 88e340256..a5986b779 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -7,7 +7,6 @@ import ( "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" -- cgit v1.2.3