aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--eth/downloader/downloader.go37
-rw-r--r--eth/handler.go18
2 files changed, 24 insertions, 31 deletions
diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 00c4cd88f..9b8d7c0b2 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -96,14 +96,14 @@ func (d *Downloader) Stats() (current int, max int) {
return d.queue.blockHashes.Size(), d.queue.fetchPool.Size() + d.queue.hashPool.Size()
}
-func (d *Downloader) RegisterPeer(id string, td *big.Int, hash common.Hash, getHashes hashFetcherFn, getBlocks blockFetcherFn) error {
+func (d *Downloader) RegisterPeer(id string, hash common.Hash, getHashes hashFetcherFn, getBlocks blockFetcherFn) error {
d.mu.Lock()
defer d.mu.Unlock()
- glog.V(logger.Detail).Infoln("Register peer", id, "TD =", td)
+ glog.V(logger.Detail).Infoln("Register peer", id)
// Create a new peer and add it to the list of known peers
- peer := newPeer(id, td, hash, getHashes, getBlocks)
+ peer := newPeer(id, hash, getHashes, getBlocks)
// add peer to our peer set
d.peers[id] = peer
// broadcast new peer
@@ -133,7 +133,7 @@ out:
break
}
- d.process()
+ d.process(peer)
case <-d.quit:
break out
}
@@ -143,27 +143,27 @@ out:
// SynchroniseWithPeer will select the peer and use it for synchronising. If an empty string is given
// it will use the best peer possible and synchronise if it's TD is higher than our own. If any of the
// checks fail an error will be returned. This method is synchronous
-func (d *Downloader) Synchronise(id string, hash common.Hash) (types.Blocks, error) {
+func (d *Downloader) Synchronise(id string, hash common.Hash) error {
// Make sure it's doing neither. Once done we can restart the
// downloading process if the TD is higher. For now just get on
// with whatever is going on. This prevents unecessary switching.
if d.isBusy() {
- return nil, errBusy
+ return errBusy
}
// Fetch the peer using the id or throw an error if the peer couldn't be found
p := d.peers[id]
if p == nil {
- return nil, errUnknownPeer
+ return errUnknownPeer
}
// Get the hash from the peer and initiate the downloading progress.
err := d.getFromPeer(p, hash, false)
if err != nil {
- return nil, err
+ return err
}
- return d.queue.blocks, nil
+ return d.process(p)
}
func (d *Downloader) getFromPeer(p *peer, hash common.Hash, ignoreInitial bool) error {
@@ -405,13 +405,12 @@ func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) error
}
peer.mu.Lock()
- peer.td = td
peer.recentHash = block.Hash()
peer.mu.Unlock()
peer.promote()
glog.V(logger.Detail).Infoln("Inserting new block from:", id)
- d.queue.addBlock(id, block, td)
+ d.queue.addBlock(id, block)
// if neither go ahead to process
if d.isBusy() {
@@ -431,10 +430,10 @@ func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) error
}
}
- return d.process()
+ return d.process(peer)
}
-func (d *Downloader) process() error {
+func (d *Downloader) process(peer *peer) error {
atomic.StoreInt32(&d.processingBlocks, 1)
defer atomic.StoreInt32(&d.processingBlocks, 0)
@@ -460,18 +459,8 @@ func (d *Downloader) process() error {
// grandparents can be requested and queued.
err = d.insertChain(blocks[:max])
if err != nil && core.IsParentErr(err) {
- glog.V(logger.Debug).Infoln("Aborting process due to missing parent. Fetching hashes")
-
- // TODO change this. This shite
- for i, block := range blocks[:max] {
- if !d.hasBlock(block.ParentHash()) {
- d.syncCh <- syncPack{d.peers.bestPeer(), block.Hash(), true}
- // remove processed blocks
- blocks = blocks[i:]
+ glog.V(logger.Debug).Infoln("Aborting process due to missing parent.")
- break
- }
- }
break
} else if err != nil {
// immediatly unregister the false peer but do not disconnect
diff --git a/eth/handler.go b/eth/handler.go
index a5bc125da..8db476eb4 100644
--- a/eth/handler.go
+++ b/eth/handler.go
@@ -90,7 +90,7 @@ type ProtocolManager struct {
minedBlockSub event.Subscription
newPeerCh chan *peer
- quit chan struct{}
+ quitSync chan struct{}
}
// NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
@@ -103,9 +103,8 @@ func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpo
downloader: downloader,
peers: make(map[string]*peer),
newPeerCh: make(chan *peer, 1),
- quit: make(chan struct{}),
+ quitSync: make(chan struct{}),
}
- go manager.peerHandler()
manager.SubProtocol = p2p.Protocol{
Name: "eth",
@@ -123,7 +122,7 @@ func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpo
return manager
}
-func (pm *ProtocolManager) peerHandler() {
+func (pm *ProtocolManager) syncHandler() {
// itimer is used to determine when to start ignoring `minDesiredPeerCount`
itimer := time.NewTimer(peerCountTimeout)
out:
@@ -153,7 +152,7 @@ out:
} else {
itimer.Reset(5 * time.Second)
}
- case <-pm.quit:
+ case <-pm.quitSync:
break out
}
}
@@ -161,7 +160,7 @@ out:
func (pm *ProtocolManager) synchronise(peer *peer) {
// Get the hashes from the peer (synchronously)
- _, err := pm.downloader.Synchronise(peer.id, peer.recentHash)
+ err := pm.downloader.Synchronise(peer.id, peer.recentHash)
if err != nil {
// handle error
glog.V(logger.Debug).Infoln("error downloading:", err)
@@ -176,11 +175,15 @@ func (pm *ProtocolManager) Start() {
// broadcast mined blocks
pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{})
go pm.minedBroadcastLoop()
+
+ // sync handler
+ go pm.syncHandler()
}
func (pm *ProtocolManager) Stop() {
pm.txSub.Unsubscribe() // quits txBroadcastLoop
pm.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop
+ close(pm.quitSync) // quits the sync handler
}
func (pm *ProtocolManager) newPeer(pv, nv int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer {
@@ -198,7 +201,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
pm.peers[p.id] = p
pm.pmu.Unlock()
- pm.downloader.RegisterPeer(p.id, p.td, p.recentHash, p.requestHashes, p.requestBlocks)
+ pm.downloader.RegisterPeer(p.id, p.recentHash, p.requestHashes, p.requestBlocks)
defer func() {
pm.pmu.Lock()
defer pm.pmu.Unlock()
@@ -370,6 +373,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
} else {
// adding blocks is synchronous
go func() {
+ // TODO check parent error
err := self.downloader.AddBlock(p.id, request.Block, request.TD)
if err != nil {
glog.V(logger.Detail).Infoln("downloader err:", err)