6 log "github.com/sirupsen/logrus"
8 "github.com/vapor/errors"
9 "github.com/vapor/netsync/peers"
10 "github.com/vapor/protocol/bc"
11 "github.com/vapor/protocol/bc/types"
15 maxNumOfSkeletonPerSync = uint64(10)
16 numOfBlocksSkeletonGap = maxNumOfBlocksPerMsg
17 maxNumOfBlocksPerSync = numOfBlocksSkeletonGap * maxNumOfSkeletonPerSync
18 fastSyncPivotGap = uint64(64)
19 minGapStartFastSync = uint64(128)
21 errNoSyncPeer = errors.New("can't find sync peer")
24 type fastSync struct {
27 blockProcessor BlockProcessor
29 mainSyncPeer *peers.Peer
32 func newFastSync(chain Chain, msgFetcher MsgFetcher, storage Storage, peers *peers.PeerSet) *fastSync {
35 msgFetcher: msgFetcher,
36 blockProcessor: newBlockProcessor(chain, storage, peers),
41 func (fs *fastSync) blockLocator() []*bc.Hash {
42 header := fs.chain.BestBlockHeader()
43 locator := []*bc.Hash{}
47 headerHash := header.Hash()
48 locator = append(locator, &headerHash)
49 if header.Height == 0 {
54 if header.Height < step {
55 header, err = fs.chain.GetHeaderByHeight(0)
57 header, err = fs.chain.GetHeaderByHeight(header.Height - step)
60 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("blockKeeper fail on get blockLocator")
64 if len(locator) >= 9 {
71 // createFetchBlocksTasks get the skeleton and assign tasks according to the skeleton.
72 func (fs *fastSync) createFetchBlocksTasks(stopBlock *types.Block) ([]*fetchBlocksWork, error) {
73 // Find peers that meet the height requirements.
74 peers := fs.peers.GetPeersByHeight(stopBlock.Height + fastSyncPivotGap)
76 return nil, errNoSyncPeer
79 // parallel fetch the skeleton from peers.
80 stopHash := stopBlock.Hash()
81 skeletonMap := fs.msgFetcher.parallelFetchHeaders(peers, fs.blockLocator(), &stopHash, numOfBlocksSkeletonGap-1)
82 if len(skeletonMap) == 0 {
83 return nil, errors.New("No skeleton found")
86 mainSkeleton, ok := skeletonMap[fs.mainSyncPeer.ID()]
88 return nil, errors.New("No main skeleton found")
91 // collect peers that match the skeleton of the primary sync peer
92 fs.msgFetcher.addSyncPeer(fs.mainSyncPeer.ID())
93 delete(skeletonMap, fs.mainSyncPeer.ID())
94 for peerID, skeleton := range skeletonMap {
95 if len(skeleton) != len(mainSkeleton) {
96 log.WithFields(log.Fields{"module": logModule, "main skeleton": len(mainSkeleton), "got skeleton": len(skeleton)}).Warn("different skeleton length")
100 for i, header := range skeleton {
101 if header.Hash() != mainSkeleton[i].Hash() {
102 log.WithFields(log.Fields{"module": logModule, "header index": i, "main skeleton": mainSkeleton[i].Hash(), "got skeleton": header.Hash()}).Warn("different skeleton hash")
106 fs.msgFetcher.addSyncPeer(peerID)
109 blockFetchTasks := make([]*fetchBlocksWork, 0)
110 // create download task
111 for i := 0; i < len(mainSkeleton)-1; i++ {
112 blockFetchTasks = append(blockFetchTasks, &fetchBlocksWork{startHeader: mainSkeleton[i], stopHeader: mainSkeleton[i+1]})
115 return blockFetchTasks, nil
118 func (fs *fastSync) process() error {
119 stopBlock, err := fs.findSyncRange()
124 tasks, err := fs.createFetchBlocksTasks(stopBlock)
129 downloadNotifyCh := make(chan struct{}, 1)
130 processStopCh := make(chan struct{})
131 var wg sync.WaitGroup
133 go fs.msgFetcher.parallelFetchBlocks(tasks, downloadNotifyCh, processStopCh, &wg)
134 go fs.blockProcessor.process(downloadNotifyCh, processStopCh, &wg)
136 fs.msgFetcher.resetParameter()
137 log.WithFields(log.Fields{"module": logModule, "height": fs.chain.BestBlockHeight()}).Info("fast sync complete")
141 // findSyncRange find the start and end of this sync.
142 // sync length cannot be greater than maxFastSyncBlocksNum.
143 func (fs *fastSync) findSyncRange() (*types.Block, error) {
144 bestHeight := fs.chain.BestBlockHeight()
145 length := fs.mainSyncPeer.IrreversibleHeight() - fastSyncPivotGap - bestHeight
146 if length > maxNumOfBlocksPerSync {
147 length = maxNumOfBlocksPerSync
150 return fs.msgFetcher.requireBlock(fs.mainSyncPeer.ID(), bestHeight+length)
153 func (fs *fastSync) setSyncPeer(peer *peers.Peer) {
154 fs.mainSyncPeer = peer