diff --git a/les/handler.go b/les/handler.go index 2189dcd471..d3601964cb 100644 --- a/les/handler.go +++ b/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 } diff --git a/les/peer.go b/les/peer.go index ea6f64a364..0c7920619a 100644 --- a/les/peer.go +++ b/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 } diff --git a/les/protocol.go b/les/protocol.go index 352833c1a0..30523c8413 100644 --- a/les/protocol.go +++ b/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 diff --git a/les/sync.go b/les/sync.go index 608e992393..c47bad19ef 100644 --- a/les/sync.go +++ b/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 { - return - } - if pm.downloader.Synchronise(peer.id, peer.Head(), peer.Td(), downloader.LightSync) != nil { + if !pm.needToSync(peer.headBlockInfo()) { 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) }