sync 3

les/handler.go

@@ -107,6 +107,9 @@ type ProtocolManager struct {
 	quitSync    chan struct{}
 	noMorePeers chan struct{}
 
+	syncMu		sync.Mutex
+	syncing		uint32
+
 	// wait group is used for graceful shutdowns during downloading
 	// and processing
 	wg sync.WaitGroup
@@ -171,7 +174,7 @@ func NewProtocolManager(chainConfig *core.ChainConfig, lightSync bool, networkId
 		manager.downloader = downloader.New(downloader.LightSync, chainDb, manager.eventMux, blockchain.HasHeader, nil, blockchain.GetHeaderByHash,
 			nil, blockchain.CurrentHeader, nil, nil, nil, blockchain.GetTdByHash,
 			blockchain.InsertHeaderChain, nil, nil, blockchain.Rollback, func(id string) {}) // manager.removePeer)
-		manager.fetcher = newLightFetcher(odr, blockchain)
+		manager.fetcher = newLightFetcher(manager)
 	}
 
 	/*validator := func(block *types.Block, parent *types.Block) error {
@@ -252,7 +255,8 @@ func (pm *ProtocolManager) handle(p *peer) error {
 
 	// Execute the LES handshake
 	td, head, genesis := pm.blockchain.Status()
-	if err := p.Handshake(td, head, genesis, pm.server); err != nil {
+	headNum := core.GetBlockNumber(pm.chainDb, head)
+	if err := p.Handshake(td, head, headNum, genesis, pm.server); err != nil {
 		glog.V(logger.Debug).Infof("%v: handshake failed: %v", p, err)
 		return err
 	}

les/peer.go

@@ -54,8 +54,8 @@ type peer struct {
 
 	id string
 
-	head common.Hash
-	td   *big.Int
+	headInfo blockInfo
+	number uint64
 	lock sync.RWMutex
 
 	knownBlocks *set.Set // Set of block hashes known to be known by this peer
@@ -92,16 +92,23 @@ func (p *peer) Head() (hash common.Hash) {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
-	copy(hash[:], p.head[:])
+	copy(hash[:], p.headInfo.Hash[:])
 	return hash
 }
 
+func (p *peer) headBlockInfo() blockInfo {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+
+	return p.headInfo
+}
+
 // Td retrieves the current total difficulty of a peer.
 func (p *peer) Td() *big.Int {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
-	return new(big.Int).Set(p.td)
+	return new(big.Int).Set(p.headInfo.Td)
 }
 
 func sendRequest(w p2p.MsgWriter, msgcode, reqID, cost uint64, data interface{}) error {
@@ -273,15 +280,16 @@ func (p *peer) sendReceiveHandshake(sendList keyValueList) (keyValueList, error)
 
 // Handshake executes the les protocol handshake, negotiating version number,
 // network IDs, difficulties, head and genesis blocks.
-func (p *peer) Handshake(td *big.Int, head common.Hash, genesis common.Hash, server *LesServer) error {
+func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, server *LesServer) error {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 
 	var send keyValueList
 	send = send.add("protocolVersion", uint64(p.version))
 	send = send.add("networkId", uint64(p.network))
-	send = send.add("td", td)
-	send = send.add("bestHash", head)
+	send = send.add("headTd", td)
+	send = send.add("headHash", head)
+	send = send.add("headNum", headNum)
 	send = send.add("genesisHash", genesis)
 	if server != nil {
 		send = send.add("serveHeaders", nil)
@@ -300,8 +308,8 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, genesis common.Hash, ser
 	}
 	recv := recvList.decode()
 	
-	var rGenesis, rBest common.Hash
-	var rVersion, rNetwork uint64
+	var rGenesis, rHash common.Hash
+	var rVersion, rNetwork, rNum uint64
 	var rTd *big.Int
 
 	if err := recv.get("protocolVersion", &rVersion); err != nil {
@@ -310,10 +318,13 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, genesis common.Hash, ser
 	if err := recv.get("networkId", &rNetwork); err != nil {
 		return err
 	}
-	if err := recv.get("td", &rTd); err != nil {
+	if err := recv.get("headTd", &rTd); err != nil {
 		return err
 	}
-	if err := recv.get("bestHash", &rBest); err != nil {
+	if err := recv.get("headHash", &rHash); err != nil {
+		return err
+	}
+	if err := recv.get("headNum", &rNum); err != nil {
 		return err
 	}
 	if err := recv.get("genesisHash", &rGenesis); err != nil {
@@ -359,7 +370,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, genesis common.Hash, ser
 		p.fcCosts = MRC.decode()
 	}
 	// Configure the remote peer, and sanity check out handshake too
-	p.td, p.head = rTd, rBest
+	p.headInfo.Td, p.headInfo.Hash, p.headInfo.Number = rTd, rHash, rNum
 	return nil
 }
 

les/protocol.go

@@ -124,8 +124,8 @@ type statusData struct {
 }
 
 // newBlockHashesData is the network packet for the block announcements.
-type newBlockHashesData []newBlockHashData
-type newBlockHashData struct {
+type newBlockHashesData []blockInfo
+type blockInfo struct {
 	Hash   common.Hash // Hash of one particular block being announced
 	Number uint64      // Number of one particular block being announced
 	Td     *big.Int    // Total difficulty of one particular block being announced

les/sync.go

@@ -17,8 +17,10 @@
 package les
 
 import (
+	"sync/atomic"
 	"time"
 
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/eth/downloader"
 )
 
@@ -56,18 +58,31 @@ func (pm *ProtocolManager) syncer() {
 	}
 }
 
+func (pm *ProtocolManager) needToSync(peerHead blockInfo) bool {
+	head := pm.blockchain.CurrentHeader()
+	currentTd := core.GetTd(pm.chainDb, head.Hash(), head.Number.Uint64())
+	return peerHead.Td.Cmp(currentTd) > 0
+}
+
 // synchronise tries to sync up our local block chain with a remote peer.
 func (pm *ProtocolManager) synchronise(peer *peer) {
 	// Short circuit if no peers are available
 	if peer == nil {
 		return
 	}
+	
 	// Make sure the peer's TD is higher than our own.
-	td := pm.blockchain.GetTdByHash(pm.blockchain.LastBlockHash())
-	if peer.Td().Cmp(td) <= 0 {
+	if !pm.needToSync(peer.headBlockInfo()) {
 		return
 	}
-	if pm.downloader.Synchronise(peer.id, peer.Head(), peer.Td(), downloader.LightSync) != nil {
-		return
+
+	pm.syncMu.Lock()
+	pm.syncWithoutLock(peer)
+	pm.syncMu.Unlock()
 }
+
+func (pm *ProtocolManager) syncWithoutLock(peer *peer) {
+	atomic.StoreUint32(&pm.syncing, 1)
+	pm.downloader.Synchronise(peer.id, peer.Head(), peer.Td(), downloader.LightSync)
+	atomic.StoreUint32(&pm.syncing, 0)
 }