以太坊区块同步详解（以太坊同步需要几天）

// filedir:go-ethereum-1.10.2\\\\eth\\\\downloader\\\\downloader.go  L96type Downloader struct {  // WARNING: The `rttEstimate` and `rttConfidence` fields are accessed atomically.  // On 32 bit platforms, only 64-bit aligned fields can be atomic. The struct is  // guaranteed to be so aligned, so take advantage of that. For more information,  // see https://golang.org/pkg/sync/atomic/#pkg-note-BUG.  rttEstimate   uint64 // Round trip time to target for download requests  rttConfidence uint64 // Confidence in the estimated RTT (unit: millionths to allow atomic ops)
  mode uint32         // Synchronisation mode defining the strategy used (per sync cycle), use d.getMode() to get the SyncMode  mux  *event.TypeMux // Event multiplexer to announce sync operation events
  checkpoint uint64   // Checkpoint block number to enforce head against (e.g. fast sync)  genesis    uint64   // Genesis block number to limit sync to (e.g. light client CHT)  queue      *queue   // Scheduler for selecting the hashes to download  peers      *peerSet // Set of active peers from which download can proceed
  stateDB    ethdb.Database  // Database to state sync into (and deduplicate via)  stateBloom *trie.SyncBloom // Bloom filter for fast trie node and contract code existence checks
  // Statistics  统计信息，  syncStatsChainOrigin uint64 // Origin block number where syncing started at  syncStatsChainHeight uint64 // Highest block number known when syncing started  syncStatsState       stateSyncStats  syncStatsLock        sync.RWMutex // Lock protecting the sync stats fields
  lightchain LightChain  blockchain BlockChain
  // Callbacks  dropPeer peerDropFn // Drops a peer for misbehaving
  // Status  synchroniseMock func(id string, hash common.Hash) error // Replacement for synchronise during testing  synchronising   int32  notified        int32  committed       int32  ancientLimit    uint64 // The maximum block number which can be regarded as ancient data.
  // Channels  headerCh      chan dataPack        // Channel receiving inbound block headers  header的输入通道，从网络下载的header会被送到这个通道  bodyCh        chan dataPack        // Channel receiving inbound block bodies   bodies的输入通道，从网络下载的bodies会被送到这个通道  receiptCh     chan dataPack        // Channel receiving inbound receipts       receipts的输入通道，从网络下载的receipts会被送到这个通道  bodyWakeCh    chan bool            // Channel to signal the block body fetcher of new tasks   用来传输body fetcher新任务的通道  receiptWakeCh chan bool            // Channel to signal the receipt fetcher of new tasks      用来传输receipt fetcher 新任务的通道  headerProcCh  chan []*types.Header // Channel to feed the header processor new tasks          通道为header处理者提供新的任务
  // State sync  pivotHeader *types.Header // Pivot block header to dynamically push the syncing state root  pivotLock   sync.RWMutex  // Lock protecting pivot header reads from updates
  snapSync       bool         // Whether to run state sync over the snap protocol  SnapSyncer     *snap.Syncer // TODO(karalabe): make private! hack for now  stateSyncStart chan *stateSync    //启动新的state fetcher  trackStateReq  chan *stateReq  stateCh        chan dataPack // Channel receiving inbound node state data  State的输入通道，从网络下载的State会被送到这个通道
  // Cancellation and termination  cancelPeer string         // Identifier of the peer currently being used as the master (cancel on drop)  cancelCh   chan struct{}  // Channel to cancel mid-flight syncs  cancelLock sync.RWMutex   // Lock to protect the cancel channel and peer in delivers  cancelWg   sync.WaitGroup // Make sure all fetcher goroutines have exited.
  quitCh   chan struct{} // Quit channel to signal termination  quitLock sync.Mutex    // Lock to prevent double closes
  // Testing hooks  syncInitHook     func(uint64, uint64)  // Method to call upon initiating a new sync run  bodyFetchHook    func([]*types.Header) // Method to call upon starting a block body fetch  receiptFetchHook func([]*types.Header) // Method to call upon starting a receipt fetch  chainInsertHook  func([]*fetchResult)  // Method to call upon inserting a chain of blocks (possibly in multiple invocations)}

// New creates a new downloader to fetch hashes and blocks from remote peers.func New(checkpoint uint64, stateDb ethdb.Database, stateBloom *trie.SyncBloom, mux *event.TypeMux, chain BlockChain, lightchain LightChain, dropPeer peerDropFn) *Downloader {  if lightchain == nil {    lightchain = chain  }  dl := &Downloader{    stateDB:        stateDb,    stateBloom:     stateBloom,    mux:            mux,    checkpoint:     checkpoint,    queue:          newQueue(blockCacheMaxItems, blockCacheInitialItems),    peers:          newPeerSet(),    rttEstimate:    uint64(rttMaxEstimate),    rttConfidence:  uint64(1000000),    blockchain:     chain,    lightchain:     lightchain,    dropPeer:       dropPeer,    headerCh:       make(chan dataPack, 1),    bodyCh:         make(chan dataPack, 1),    receiptCh:      make(chan dataPack, 1),    bodyWakeCh:     make(chan bool, 1),    receiptWakeCh:  make(chan bool, 1),    headerProcCh:   make(chan []*types.Header, 1),    quitCh:         make(chan struct{}),    stateCh:        make(chan dataPack),    SnapSyncer:     snap.NewSyncer(stateDb),    stateSyncStart: make(chan *stateSync),    syncStatsState: stateSyncStats{      processed: rawdb.ReadFastTrieProgress(stateDb),    },    trackStateReq: make(chan *stateReq),  }  go dl.qosTuner()        //计算rttEstimate和rttConfidence  go dl.stateFetcher()    //启动stateFetcher的任务监听  return dl}

// Synchronise tries to sync up our local block chain with a remote peer, both// adding various sanity checks as well as wrapping it with various log entries.func (d *Downloader) Synchronise(id string, head common.Hash, td *big.Int, mode SyncMode) error {  err := d.synchronise(id, head, td, mode)
  switch err {  case nil, errBusy, errCanceled:    return err  }  if errors.Is(err, errInvalidChain) || errors.Is(err, errBadPeer) || errors.Is(err, errTimeout) ||    errors.Is(err, errStallingPeer) || errors.Is(err, errUnsyncedPeer) || errors.Is(err, errEmptyHeaderSet) ||    errors.Is(err, errPeersUnavailable) || errors.Is(err, errTooOld) || errors.Is(err, errInvalidAncestor) {    log.Warn(\\\"Synchronisation failed, dropping peer\\\", \\\"peer\\\", id, \\\"err\\\", err)    if d.dropPeer == nil {      // The dropPeer method is nil when `--copydb` is used for a local copy.      // Timeouts can occur if e.g. compaction hits at the wrong time, and can be ignored      log.Warn(\\\"Downloader wants to drop peer, but peerdrop-function is not set\\\", \\\"peer\\\", id)    } else {      d.dropPeer(id)    }    return err  }  log.Warn(\\\"Synchronisation failed, retrying\\\", \\\"err\\\", err)  return err}

// synchronise will select the peer and use it for synchronising. If an empty string is given// it will use the best peer possible and synchronize if its TD is higher than our own. If any of the// checks fail an error will be returned. This method is synchronousfunc (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode SyncMode) error {  // Mock out the synchronisation if testing  if d.synchroniseMock != nil {    return d.synchroniseMock(id, hash)  }  // Make sure only one goroutine is ever allowed past this point at once    // 只能运行一个， 检查是否正在运行  if !atomic.CompareAndSwapInt32(&d.synchronising, 0, 1) {    return errBusy  }  defer atomic.StoreInt32(&d.synchronising, 0)
  // Post a user notification of the sync (only once per session)   // 发布同步的用户通知（每个会话仅一次）  if atomic.CompareAndSwapInt32(&d.notified, 0, 1) {    log.Info(\\\"Block synchronisation started\\\")  }  // If we are already full syncing, but have a fast-sync bloom filter laying  // around, make sure it doesn\\\'t use memory any more. This is a special case  // when the user attempts to fast sync a new empty network.  if mode == FullSync && d.stateBloom != nil {    d.stateBloom.Close()  }  // If snap sync was requested, create the snap scheduler and switch to fast  // sync mode. Long term we could drop fast sync or merge the two together,  // but until snap becomes prevalent, we should support both. TODO(karalabe).  if mode == SnapSync {    if !d.snapSync {      log.Warn(\\\"Enabling snapshot sync prototype\\\")      d.snapSync = true    }    mode = FastSync  }  // Reset the queue, peer set and wake channels to clean any internal leftover state    d.queue.Reset(blockCacheMaxItems, blockCacheInitialItems)  // 重置queue的状态  d.peers.Reset()                       //  重置peer的状态
  for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {   // 清空d.bodyWakeCh, d.receiptWakeCh    select {    case <-ch:    default:    }  }  for _, ch := range []chan dataPack{d.headerCh, d.bodyCh, d.receiptCh} {   //清空d.headerCh, d.bodyCh, d.receiptCh    for empty := false; !empty; {      select {      case <-ch:      default:        empty = true      }    }  }  for empty := false; !empty; {    // 清空headerProcCh    select {    case <-d.headerProcCh:    default:      empty = true    }  }  // Create cancel channel for aborting mid-flight and mark the master peer  d.cancelLock.Lock()  d.cancelCh = make(chan struct{})  d.cancelPeer = id  d.cancelLock.Unlock()
  defer d.Cancel() // No matter what, we can\\\'t leave the cancel channel open
  // Atomically set the requested sync mode  atomic.StoreUint32(&d.mode, uint32(mode))
  // Retrieve the origin peer and initiate the downloading process  p := d.peers.Peer(id)  if p == nil {    return errUnknownPeer  }  return d.syncWithPeer(p, hash, td)    // 基于哈希链从指定的peer和head hash开始块同步}

// filedir:go-ethereum-1.10.2\\\\eth\\\\downloader\\\\downloader.go  L448// syncWithPeer starts a block synchronization based on the hash chain from the// specified peer and head hash.func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td *big.Int) (err error) {  d.mux.Post(StartEvent{})  defer func() {    // reset on error    if err != nil {      d.mux.Post(FailedEvent{err})    } else {      latest := d.lightchain.CurrentHeader()      d.mux.Post(DoneEvent{latest})    }  }()  if p.version < 64 {    return fmt.Errorf(\\\"%w: advertized %d < required %d\\\", errTooOld, p.version, 64)  }  mode := d.getMode()
  log.Debug(\\\"Synchronising with the network\\\", \\\"peer\\\", p.id, \\\"eth\\\", p.version, \\\"head\\\", hash, \\\"td\\\", td, \\\"mode\\\", mode)  defer func(start time.Time) {    log.Debug(\\\"Synchronisation terminated\\\", \\\"elapsed\\\", common.PrettyDuration(time.Since(start)))  }(time.Now())
  // Look up the sync boundaries: the common ancestor and the target block  latest, pivot, err := d.fetchHead(p)     if err != nil {    return err  }  if mode == FastSync && pivot == nil {    // If no pivot block was returned, the head is below the min full block    // threshold (i.e. new chian). In that case we won\\\'t really fast sync    // anyway, but still need a valid pivot block to avoid some code hitting    // nil panics on an access.    pivot = d.blockchain.CurrentBlock().Header()  }  height := latest.Number.Uint64()
  origin, err := d.findAncestor(p, latest)  // 通过findAncestor来获取共同祖先，以便找到一个开始同步的点  if err != nil {    return err  }  d.syncStatsLock.Lock()  if d.syncStatsChainHeight <= origin || d.syncStatsChainOrigin > origin {    d.syncStatsChainOrigin = origin  }  d.syncStatsChainHeight = height  d.syncStatsLock.Unlock()
  // Ensure our origin point is below any fast sync pivot point  if mode == FastSync {    if height <= uint64(fsMinFullBlocks) {      // 如果对端节点的height小于64，则共同祖先更新为0      origin = 0    } else {                   // 否则更新pivot为对端节点height-64      pivotNumber := pivot.Number.Uint64()      if pivotNumber <= origin {             // 如果pivot小于共同祖先，则更新共同祖先为pivot的前一个        origin = pivotNumber - 1      }      // Write out the pivot into the database so a rollback beyond it will      // reenable fast sync      rawdb.WriteLastPivotNumber(d.stateDB, pivotNumber)    }  }  d.committed = 1  if mode == FastSync && pivot.Number.Uint64() != 0 {    d.committed = 0  }  if mode == FastSync {    // Set the ancient data limitation.    // If we are running fast sync, all block data older than ancientLimit will be    // written to the ancient store. More recent data will be written to the active    // database and will wait for the freezer to migrate.    //    // If there is a checkpoint available, then calculate the ancientLimit through    // that. Otherwise calculate the ancient limit through the advertised height    // of the remote peer.    //    // The reason for picking checkpoint first is that a malicious peer can give us    // a fake (very high) height, forcing the ancient limit to also be very high.    // The peer would start to feed us valid blocks until head, resulting in all of    // the blocks might be written into the ancient store. A following mini-reorg    // could cause issues.    if d.checkpoint != 0 && d.checkpoint > fullMaxForkAncestry+1 {      d.ancientLimit = d.checkpoint    } else if height > fullMaxForkAncestry+1 {      d.ancientLimit = height - fullMaxForkAncestry - 1    } else {      d.ancientLimit = 0    }    frozen, _ := d.stateDB.Ancients() // Ignore the error here since light client can also hit here.
    // If a part of blockchain data has already been written into active store,    // disable the ancient style insertion explicitly.    if origin >= frozen && frozen != 0 {      d.ancientLimit = 0      log.Info(\\\"Disabling direct-ancient mode\\\", \\\"origin\\\", origin, \\\"ancient\\\", frozen-1)    } else if d.ancientLimit > 0 {      log.Debug(\\\"Enabling direct-ancient mode\\\", \\\"ancient\\\", d.ancientLimit)    }    // Rewind the ancient store and blockchain if reorg happens.    if origin+1 < frozen {      if err := d.lightchain.SetHead(origin + 1); err != nil {        return err      }    }  }  // Initiate the sync using a concurrent header and content retrieval algorithm  d.queue.Prepare(origin+1, mode) // 更新queue的值从共同祖先+1开始，即从共同祖先开始sync区块  if d.syncInitHook != nil {    d.syncInitHook(origin, height)  }  fetchers := []func() error{    func() error { return d.fetchHeaders(p, origin+1) }, // Headers are always retrieved    func() error { return d.fetchBodies(origin + 1) },   // Bodies are retrieved during normal and fast sync    func() error { return d.fetchReceipts(origin + 1) }, // Receipts are retrieved during fast sync    func() error { return d.processHeaders(origin+1, td) },  }  if mode == FastSync {        //根据模式的不同，增加新的处理逻辑    d.pivotLock.Lock()    d.pivotHeader = pivot    d.pivotLock.Unlock()
    fetchers = append(fetchers, func() error { return d.processFastSyncContent() })  } else if mode == FullSync {    fetchers = append(fetchers, d.processFullSyncContent)  }  return d.spawnSync(fetchers)}

// spawnSync runs d.process and all given fetcher functions to completion in// separate goroutines, returning the first error that appears.func (d *Downloader) spawnSync(fetchers []func() error) error {  errc := make(chan error, len(fetchers))  d.cancelWg.Add(len(fetchers))  for _, fn := range fetchers {    fn := fn    go func() { defer d.cancelWg.Done(); errc <- fn() }()  }  // Wait for the first error, then terminate the others.  var err error  for i := 0; i < len(fetchers); i++ {    if i == len(fetchers)-1 {      // Close the queue when all fetchers have exited.      // This will cause the block processor to end when      // it has processed the queue.      d.queue.Close()    }    if err = <-errc; err != nil && err != errCanceled {      break    }  }  d.queue.Close()  d.Cancel()  return err}

// filedir: go-ethereum-1.10.2\\\\eth\\\\downloader\\\\statesync.go// stateFetcher manages the active state sync and accepts requests// on its behalf.func (d *Downloader) stateFetcher() {  for {    select {    case s := <-d.stateSyncStart:      for next := s; next != nil; {        next = d.runStateSync(next)      }    case <-d.stateCh:      // Ignore state responses while no sync is running.    case <-d.quitCh:      return    }  }}

// runStateSync runs a state synchronisation until it completes or another root// hash is requested to be switched over to.func (d *Downloader) runStateSync(s *stateSync) *stateSync {  var (    active   = make(map[string]*stateReq) // Currently in-flight requests    finished []*stateReq                  // Completed or failed requests    timeout  = make(chan *stateReq)       // Timed out active requests  )  log.Trace(\\\"State sync starting\\\", \\\"root\\\", s.root)
  defer func() {    // Cancel active request timers on exit. Also set peers to idle so they\\\'re    // available for the next sync.    for _, req := range active {      req.timer.Stop()      req.peer.SetNodeDataIdle(int(req.nItems), time.Now())    }  }()  go s.run()  defer s.Cancel()
  // Listen for peer departure events to cancel assigned tasks  peerDrop := make(chan *peerConnection, 1024)  peerSub := s.d.peers.SubscribePeerDrops(peerDrop)  defer peerSub.Unsubscribe()
  for {    // Enable sending of the first buffered element if there is one.    var (      deliverReq   *stateReq      deliverReqCh chan *stateReq    )    if len(finished) > 0 {      deliverReq = finished[0]      deliverReqCh = s.deliver    }
    select {    // The stateSync lifecycle:    case next := <-d.stateSyncStart:      d.spindownStateSync(active, finished, timeout, peerDrop)      return next
    case <-s.done:      d.spindownStateSync(active, finished, timeout, peerDrop)      return nil
    // Send the next finished request to the current sync:    case deliverReqCh <- deliverReq:      // Shift out the first request, but also set the emptied slot to nil for GC      copy(finished, finished[1:])      finished[len(finished)-1] = nil      finished = finished[:len(finished)-1]
    // Handle incoming state packs:    case pack := <-d.stateCh:      // Discard any data not requested (or previously timed out)      req := active[pack.PeerId()]      if req == nil {        log.Debug(\\\"Unrequested node data\\\", \\\"peer\\\", pack.PeerId(), \\\"len\\\", pack.Items())        continue      }      // Finalize the request and queue up for processing      req.timer.Stop()      req.response = pack.(*statePack).states      req.delivered = time.Now()
      finished = append(finished, req)      delete(active, pack.PeerId())
    // Handle dropped peer connections:    case p := <-peerDrop:      // Skip if no request is currently pending      req := active[p.id]      if req == nil {        continue      }      // Finalize the request and queue up for processing      req.timer.Stop()      req.dropped = true      req.delivered = time.Now()
      finished = append(finished, req)      delete(active, p.id)
    // Handle timed-out requests:    case req := <-timeout:      // If the peer is already requesting something else, ignore the stale timeout.      // This can happen when the timeout and the delivery happens simultaneously,      // causing both pathways to trigger.      if active[req.peer.id] != req {        continue      }      req.delivered = time.Now()      // Move the timed out data back into the download queue      finished = append(finished, req)      delete(active, req.peer.id)
    // Track outgoing state requests:    case req := <-d.trackStateReq:      // If an active request already exists for this peer, we have a problem. In      // theory the trie node schedule must never assign two requests to the same      // peer. In practice however, a peer might receive a request, disconnect and      // immediately reconnect before the previous times out. In this case the first      // request is never honored, alas we must not silently overwrite it, as that      // causes valid requests to go missing and sync to get stuck.      if old := active[req.peer.id]; old != nil {        log.Warn(\\\"Busy peer assigned new state fetch\\\", \\\"peer\\\", old.peer.id)        // Move the previous request to the finished set        old.timer.Stop()        old.dropped = true        old.delivered = time.Now()        finished = append(finished, old)      }      // Start a timer to notify the sync loop if the peer stalled.      req.timer = time.AfterFunc(req.timeout, func() {        timeout <- req      })      active[req.peer.id] = req    }  }}

// fetchHead retrieves the head header and prior pivot block (if available) from// a remote peer.func (d *Downloader) fetchHead(p *peerConnection) (head *types.Header, pivot *types.Header, err error) {  p.log.Debug(\\\"Retrieving remote chain head\\\")  mode := d.getMode()
  // Request the advertised remote head block and wait for the response  latest, _ := p.peer.Head()  fetch := 1  if mode == FastSync {    fetch = 2 // head + pivot headers  }  go p.peer.RequestHeadersByHash(latest, fetch, fsMinFullBlocks-1, true)
  ttl := d.requestTTL()  timeout := time.After(ttl)  for {    select {    case <-d.cancelCh:      return nil, nil, errCanceled
    case packet := <-d.headerCh:      // Discard anything not from the origin peer      if packet.PeerId() != p.id {        log.Debug(\\\"Received headers from incorrect peer\\\", \\\"peer\\\", packet.PeerId())        break      }      // Make sure the peer gave us at least one and at most the requested headers      headers := packet.(*headerPack).headers      if len(headers) == 0 || len(headers) > fetch {        return nil, nil, fmt.Errorf(\\\"%w: returned headers %d != requested %d\\\", errBadPeer, len(headers), fetch)      }      // The first header needs to be the head, validate against the checkpoint      // and request. If only 1 header was returned, make sure there\\\'s no pivot      // or there was not one requested.      head := headers[0]      if (mode == FastSync || mode == LightSync) && head.Number.Uint64() < d.checkpoint {        return nil, nil, fmt.Errorf(\\\"%w: remote head %d below checkpoint %d\\\", errUnsyncedPeer, head.Number, d.checkpoint)      }      if len(headers) == 1 {        if mode == FastSync && head.Number.Uint64() > uint64(fsMinFullBlocks) {          return nil, nil, fmt.Errorf(\\\"%w: no pivot included along head header\\\", errBadPeer)        }        p.log.Debug(\\\"Remote head identified, no pivot\\\", \\\"number\\\", head.Number, \\\"hash\\\", head.Hash())        return head, nil, nil      }      // At this point we have 2 headers in total and the first is the      // validated head of the chian. Check the pivot number and return,      pivot := headers[1]      if pivot.Number.Uint64() != head.Number.Uint64()-uint64(fsMinFullBlocks) {        return nil, nil, fmt.Errorf(\\\"%w: remote pivot %d != requested %d\\\", errInvalidChain, pivot.Number, head.Number.Uint64()-uint64(fsMinFullBlocks))      }      return head, pivot, nil
    case <-timeout:      p.log.Debug(\\\"Waiting for head header timed out\\\", \\\"elapsed\\\", ttl)      return nil, nil, errTimeout
    case <-d.bodyCh:    case <-d.receiptCh:      // Out of bounds delivery, ignore    }  }}

// processHeaders takes batches of retrieved headers from an input channel and// keeps processing and scheduling them into the header chain and downloader\\\'s// queue until the stream ends or a failure occurs.func (d *Downloader) processHeaders(origin uint64, td *big.Int) error {  // Keep a count of uncertain headers to roll back  var (    rollback    uint64 // Zero means no rollback (fine as you can\\\'t unroll the genesis)    rollbackErr error    mode        = d.getMode()  )  defer func() {    if rollback > 0 {      lastHeader, lastFastBlock, lastBlock := d.lightchain.CurrentHeader().Number, common.Big0, common.Big0      if mode != LightSync {        lastFastBlock = d.blockchain.CurrentFastBlock().Number()        lastBlock = d.blockchain.CurrentBlock().Number()      }      if err := d.lightchain.SetHead(rollback - 1); err != nil { // -1 to target the parent of the first uncertain block        // We\\\'re already unwinding the stack, only print the error to make it more visible        log.Error(\\\"Failed to roll back chain segment\\\", \\\"head\\\", rollback-1, \\\"err\\\", err)      }      curFastBlock, curBlock := common.Big0, common.Big0      if mode != LightSync {        curFastBlock = d.blockchain.CurrentFastBlock().Number()        curBlock = d.blockchain.CurrentBlock().Number()      }      log.Warn(\\\"Rolled back chain segment\\\",        \\\"header\\\", fmt.Sprintf(\\\"%d->%d\\\", lastHeader, d.lightchain.CurrentHeader().Number),        \\\"fast\\\", fmt.Sprintf(\\\"%d->%d\\\", lastFastBlock, curFastBlock),        \\\"block\\\", fmt.Sprintf(\\\"%d->%d\\\", lastBlock, curBlock), \\\"reason\\\", rollbackErr)    }  }()  // Wait for batches of headers to process  gotHeaders := false
  for {    select {    case <-d.cancelCh:      rollbackErr = errCanceled      return errCanceled
    case headers := <-d.headerProcCh:      // Terminate header processing if we synced up      if len(headers) == 0 {        // Notify everyone that headers are fully processed        for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {          select {          case ch <- false:          case <-d.cancelCh:          }        }        // If no headers were retrieved at all, the peer violated its TD promise that it had a        // better chain compared to ours. The only exception is if its promised blocks were        // already imported by other means (e.g. fetcher):        //        // R <remote peer>, L <local node>: Both at block 10        // R: Mine block 11, and propagate it to L        // L: Queue block 11 for import        // L: Notice that R\\\'s head and TD increased compared to ours, start sync        // L: Import of block 11 finishes        // L: Sync begins, and finds common ancestor at 11        // L: Request new headers up from 11 (R\\\'s TD was higher, it must have something)        // R: Nothing to give        if mode != LightSync {          head := d.blockchain.CurrentBlock()          if !gotHeaders && td.Cmp(d.blockchain.GetTd(head.Hash(), head.NumberU64())) > 0 {            return errStallingPeer          }        }        // If fast or light syncing, ensure promised headers are indeed delivered. This is        // needed to detect scenarios where an attacker feeds a bad pivot and then bails out        // of delivering the post-pivot blocks that would flag the invalid content.        //        // This check cannot be executed \\\"as is\\\" for full imports, since blocks may still be        // queued for processing when the header download completes. However, as long as the        // peer gave us something useful, we\\\'re already happy/progressed (above check).        if mode == FastSync || mode == LightSync {          head := d.lightchain.CurrentHeader()          if td.Cmp(d.lightchain.GetTd(head.Hash(), head.Number.Uint64())) > 0 {            return errStallingPeer          }        }        // Disable any rollback and return        rollback = 0        return nil      }      // Otherwise split the chunk of headers into batches and process them      gotHeaders = true      for len(headers) > 0 {        // Terminate if something failed in between processing chunks        select {        case <-d.cancelCh:          rollbackErr = errCanceled          return errCanceled        default:        }        // Select the next chunk of headers to import        limit := maxHeadersProcess        if limit > len(headers) {          limit = len(headers)        }        chunk := headers[:limit]
        // In case of header only syncing, validate the chunk immediately        if mode == FastSync || mode == LightSync {          // If we\\\'re importing pure headers, verify based on their recentness          var pivot uint64
          d.pivotLock.RLock()          if d.pivotHeader != nil {            pivot = d.pivotHeader.Number.Uint64()          }          d.pivotLock.RUnlock()
          frequency := fsHeaderCheckFrequency          if chunk[len(chunk)-1].Number.Uint64()+uint64(fsHeaderForceVerify) > pivot {            frequency = 1          }          if n, err := d.lightchain.InsertHeaderChain(chunk, frequency); err != nil {            rollbackErr = err
            // If some headers were inserted, track them as uncertain            if (mode == FastSync || frequency > 1) && n > 0 && rollback == 0 {              rollback = chunk[0].Number.Uint64()            }            log.Warn(\\\"Invalid header encountered\\\", \\\"number\\\", chunk[n].Number, \\\"hash\\\", chunk[n].Hash(), \\\"parent\\\", chunk[n].ParentHash, \\\"err\\\", err)            return fmt.Errorf(\\\"%w: %v\\\", errInvalidChain, err)          }          // All verifications passed, track all headers within the alloted limits          if mode == FastSync {            head := chunk[len(chunk)-1].Number.Uint64()            if head-rollback > uint64(fsHeaderSafetyNet) {              rollback = head - uint64(fsHeaderSafetyNet)            } else {              rollback = 1            }          }        }        // Unless we\\\'re doing light chains, schedule the headers for associated content retrieval        if mode == FullSync || mode == FastSync {          // If we\\\'ve reached the allowed number of pending headers, stall a bit          for d.queue.PendingBlocks() >= maxQueuedHeaders || d.queue.PendingReceipts() >= maxQueuedHeaders {            select {            case <-d.cancelCh:              rollbackErr = errCanceled              return errCanceled            case <-time.After(time.Second):            }          }          // Otherwise insert the headers for content retrieval          inserts := d.queue.Schedule(chunk, origin)          if len(inserts) != len(chunk) {            rollbackErr = fmt.Errorf(\\\"stale headers: len inserts %v len(chunk) %v\\\", len(inserts), len(chunk))            return fmt.Errorf(\\\"%w: stale headers\\\", errBadPeer)          }        }        headers = headers[limit:]        origin += uint64(limit)      }      // Update the highest block number we know if a higher one is found.      d.syncStatsLock.Lock()      if d.syncStatsChainHeight < origin {        d.syncStatsChainHeight = origin - 1      }      d.syncStatsLock.Unlock()
      // Signal the content downloaders of the availablility of new tasks      for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {        select {        case ch <- true:        default:        }      }    }  }}

// fetchBodies iteratively downloads the scheduled block bodies, taking any// available peers, reserving a chunk of blocks for each, waiting for delivery// and also periodically checking for timeouts.func (d *Downloader) fetchBodies(from uint64) error {  log.Debug(\\\"Downloading block bodies\\\", \\\"origin\\\", from)
  var (    deliver = func(packet dataPack) (int, error) {      pack := packet.(*bodyPack)      return d.queue.DeliverBodies(pack.peerID, pack.transactions, pack.uncles)    }    expire   = func() map[string]int { return d.queue.ExpireBodies(d.requestTTL()) }    fetch    = func(p *peerConnection, req *fetchRequest) error { return p.FetchBodies(req) }    capacity = func(p *peerConnection) int { return p.BlockCapacity(d.requestRTT()) }    setIdle  = func(p *peerConnection, accepted int, deliveryTime time.Time) { p.SetBodiesIdle(accepted, deliveryTime) }  )  err := d.fetchParts(d.bodyCh, deliver, d.bodyWakeCh, expire,    d.queue.PendingBlocks, d.queue.InFlightBlocks, d.queue.ReserveBodies,    d.bodyFetchHook, fetch, d.queue.CancelBodies, capacity, d.peers.BodyIdlePeers, setIdle, \\\"bodies\\\")
  log.Debug(\\\"Block body download terminated\\\", \\\"err\\\", err)  return err}

// DeliverBodies injects a new batch of block bodies received from a remote node.func (d *Downloader) DeliverBodies(id string, transactions [][]*types.Transaction, uncles [][]*types.Header) error {  return d.deliver(d.bodyCh, &bodyPack{id, transactions, uncles}, bodyInMeter, bodyDropMeter)}

// deliver injects a new batch of data received from a remote node.func (d *Downloader) deliver(destCh chan dataPack, packet dataPack, inMeter, dropMeter metrics.Meter) (err error) {  // Update the delivery metrics for both good and failed deliveries  inMeter.Mark(int64(packet.Items()))  defer func() {    if err != nil {      dropMeter.Mark(int64(packet.Items()))    }  }()  // Deliver or abort if the sync is canceled while queuing  d.cancelLock.RLock()  cancel := d.cancelCh  d.cancelLock.RUnlock()  if cancel == nil {    return errNoSyncActive  }  select {  case destCh <- packet:    return nil  case <-cancel:    return errNoSyncActive  }}

func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack) (int, error), wakeCh chan bool,  expire func() map[string]int, pending func() int, inFlight func() bool, reserve func(*peerConnection, int) (*fetchRequest, bool, bool),  fetchHook func([]*types.Header), fetch func(*peerConnection, *fetchRequest) error, cancel func(*fetchRequest), capacity func(*peerConnection) int,  idle func() ([]*peerConnection, int), setIdle func(*peerConnection, int, time.Time), kind string) error {
  // Create a ticker to detect expired retrieval tasks  ticker := time.NewTicker(100 * time.Millisecond)  defer ticker.Stop()
  update := make(chan struct{}, 1)
  // Prepare the queue and fetch block parts until the block header fetcher\\\'s done  finished := false  for {    select {    case <-d.cancelCh:      return errCanceled
    case packet := <-deliveryCh:      deliveryTime := time.Now()      // If the peer was previously banned and failed to deliver its pack      // in a reasonable time frame, ignore its message.      if peer := d.peers.Peer(packet.PeerId()); peer != nil {        // Deliver the received chunk of data and check chain validity        accepted, err := deliver(packet)        if errors.Is(err, errInvalidChain) {          return err        }        // Unless a peer delivered something completely else than requested (usually        // caused by a timed out request which came through in the end), set it to        // idle. If the delivery\\\'s stale, the peer should have already been idled.        if !errors.Is(err, errStaleDelivery) {          setIdle(peer, accepted, deliveryTime)        }        // Issue a log to the user to see what\\\'s going on        switch {        case err == nil && packet.Items() == 0:          peer.log.Trace(\\\"Requested data not delivered\\\", \\\"type\\\", kind)        case err == nil:          peer.log.Trace(\\\"Delivered new batch of data\\\", \\\"type\\\", kind, \\\"count\\\", packet.Stats())        default:          peer.log.Debug(\\\"Failed to deliver retrieved data\\\", \\\"type\\\", kind, \\\"err\\\", err)        }      }      // Blocks assembled, try to update the progress      select {      case update <- struct{}{}:      default:      }
    case cont := <-wakeCh:      // The header fetcher sent a continuation flag, check if it\\\'s done      if !cont {        finished = true      }      // Headers arrive, try to update the progress      select {      case update <- struct{}{}:      default:      }
    case <-ticker.C:      // Sanity check update the progress      select {      case update <- struct{}{}:      default:      }
    case <-update:      // Short circuit if we lost all our peers      if d.peers.Len() == 0 {        return errNoPeers      }      // Check for fetch request timeouts and demote the responsible peers      for pid, fails := range expire() {        if peer := d.peers.Peer(pid); peer != nil {          // If a lot of retrieval elements expired, we might have overestimated the remote peer or perhaps          // ourselves. Only reset to minimal throughput but don\\\'t drop just yet. If even the minimal times          // out that sync wise we need to get rid of the peer.          //          // The reason the minimum threshold is 2 is because the downloader tries to estimate the bandwidth          // and latency of a peer separately, which requires pushing the measures capacity a bit and seeing          // how response times reacts, to it always requests one more than the minimum (i.e. min 2).          if fails > 2 {            peer.log.Trace(\\\"Data delivery timed out\\\", \\\"type\\\", kind)            setIdle(peer, 0, time.Now())          } else {            peer.log.Debug(\\\"Stalling delivery, dropping\\\", \\\"type\\\", kind)
            if d.dropPeer == nil {              // The dropPeer method is nil when `--copydb` is used for a local copy.              // Timeouts can occur if e.g. compaction hits at the wrong time, and can be ignored              peer.log.Warn(\\\"Downloader wants to drop peer, but peerdrop-function is not set\\\", \\\"peer\\\", pid)            } else {              d.dropPeer(pid)
              // If this peer was the master peer, abort sync immediately              d.cancelLock.RLock()              master := pid == d.cancelPeer              d.cancelLock.RUnlock()
              if master {                d.cancel()                return errTimeout              }            }          }        }      }      // If there\\\'s nothing more to fetch, wait or terminate      if pending() == 0 {        if !inFlight() && finished {          log.Debug(\\\"Data fetching completed\\\", \\\"type\\\", kind)          return nil        }        break      }      // Send a download request to all idle peers, until throttled      progressed, throttled, running := false, false, inFlight()      idles, total := idle()      pendCount := pending()      for _, peer := range idles {        // Short circuit if throttling activated        if throttled {          break        }        // Short circuit if there is no more available task.        if pendCount = pending(); pendCount == 0 {          break        }        // Reserve a chunk of fetches for a peer. A nil can mean either that        // no more headers are available, or that the peer is known not to        // have them.        request, progress, throttle := reserve(peer, capacity(peer))        if progress {          progressed = true        }        if throttle {          throttled = true          throttleCounter.Inc(1)        }        if request == nil {          continue        }        if request.From > 0 {          peer.log.Trace(\\\"Requesting new batch of data\\\", \\\"type\\\", kind, \\\"from\\\", request.From)        } else {          peer.log.Trace(\\\"Requesting new batch of data\\\", \\\"type\\\", kind, \\\"count\\\", len(request.Headers), \\\"from\\\", request.Headers[0].Number)        }        // Fetch the chunk and make sure any errors return the hashes to the queue        if fetchHook != nil {          fetchHook(request.Headers)        }        if err := fetch(peer, request); err != nil {          // Although we could try and make an attempt to fix this, this error really          // means that we\\\'ve double allocated a fetch task to a peer. If that is the          // case, the internal state of the downloader and the queue is very wrong so          // better hard crash and note the error instead of silently accumulating into          // a much bigger issue.          panic(fmt.Sprintf(\\\"%v: %s fetch assignment failed\\\", peer, kind))        }        running = true      }      // Make sure that we have peers available for fetching. If all peers have been tried      // and all failed throw an error      if !progressed && !throttled && !running && len(idles) == total && pendCount > 0 {        return errPeersUnavailable      }    }  }}

// fetchReceipts iteratively downloads the scheduled block receipts, taking any// available peers, reserving a chunk of receipts for each, waiting for delivery// and also periodically checking for timeouts.func (d *Downloader) fetchReceipts(from uint64) error {  log.Debug(\\\"Downloading transaction receipts\\\", \\\"origin\\\", from)
  var (    deliver = func(packet dataPack) (int, error) {      pack := packet.(*receiptPack)      return d.queue.DeliverReceipts(pack.peerID, pack.receipts)    }    expire   = func() map[string]int { return d.queue.ExpireReceipts(d.requestTTL()) }    fetch    = func(p *peerConnection, req *fetchRequest) error { return p.FetchReceipts(req) }    capacity = func(p *peerConnection) int { return p.ReceiptCapacity(d.requestRTT()) }    setIdle  = func(p *peerConnection, accepted int, deliveryTime time.Time) {      p.SetReceiptsIdle(accepted, deliveryTime)    }  )  err := d.fetchParts(d.receiptCh, deliver, d.receiptWakeCh, expire,    d.queue.PendingReceipts, d.queue.InFlightReceipts, d.queue.ReserveReceipts,    d.receiptFetchHook, fetch, d.queue.CancelReceipts, capacity, d.peers.ReceiptIdlePeers, setIdle, \\\"receipts\\\")
  log.Debug(\\\"Transaction receipt download terminated\\\", \\\"err\\\", err)  return err}

// DeliverReceipts injects a new batch of receipts received from a remote node.func (d *Downloader) DeliverReceipts(id string, receipts [][]*types.Receipt) error {  return d.deliver(d.receiptCh, &receiptPack{id, receipts}, receiptInMeter, receiptDropMeter)}

// deliver injects a new batch of data received from a remote node.func (d *Downloader) deliver(destCh chan dataPack, packet dataPack, inMeter, dropMeter metrics.Meter) (err error) {  // Update the delivery metrics for both good and failed deliveries  inMeter.Mark(int64(packet.Items()))  defer func() {    if err != nil {      dropMeter.Mark(int64(packet.Items()))    }  }()  // Deliver or abort if the sync is canceled while queuing  d.cancelLock.RLock()  cancel := d.cancelCh  d.cancelLock.RUnlock()  if cancel == nil {    return errNoSyncActive  }  select {  case destCh <- packet:    return nil  case <-cancel:    return errNoSyncActive  }}

// processFullSyncContent takes fetch results from the queue and imports them into the chain.func (d *Downloader) processFullSyncContent() error {  for {    results := d.queue.Results(true)    if len(results) == 0 {      return nil    }    if d.chainInsertHook != nil {      d.chainInsertHook(results)    }    if err := d.importBlockResults(results); err != nil {      return err    }  }}

// processFastSyncContent takes fetch results from the queue and writes them to the// database. It also controls the synchronisation of state nodes of the pivot block.func (d *Downloader) processFastSyncContent() error {  // Start syncing state of the reported head block. This should get us most of  // the state of the pivot block.  d.pivotLock.RLock()  sync := d.syncState(d.pivotHeader.Root)  d.pivotLock.RUnlock()
  defer func() {    // The `sync` object is replaced every time the pivot moves. We need to    // defer close the very last active one, hence the lazy evaluation vs.    // calling defer sync.Cancel() !!!    sync.Cancel()  }()
  closeOnErr := func(s *stateSync) {    if err := s.Wait(); err != nil && err != errCancelStateFetch && err != errCanceled && err != snap.ErrCancelled {      d.queue.Close() // wake up Results    }  }  go closeOnErr(sync)
  // To cater for moving pivot points, track the pivot block and subsequently  // accumulated download results separately.  var (    oldPivot *fetchResult   // Locked in pivot block, might change eventually    oldTail  []*fetchResult // Downloaded content after the pivot  )  for {    // Wait for the next batch of downloaded data to be available, and if the pivot    // block became stale, move the goalpost    results := d.queue.Results(oldPivot == nil) // Block if we\\\'re not monitoring pivot staleness    if len(results) == 0 {      // If pivot sync is done, stop      if oldPivot == nil {        return sync.Cancel()      }      // If sync failed, stop      select {      case <-d.cancelCh:        sync.Cancel()        return errCanceled      default:      }    }    if d.chainInsertHook != nil {      d.chainInsertHook(results)    }    // If we haven\\\'t downloaded the pivot block yet, check pivot staleness    // notifications from the header downloader    d.pivotLock.RLock()    pivot := d.pivotHeader    d.pivotLock.RUnlock()
    if oldPivot == nil {      if pivot.Root != sync.root {        sync.Cancel()        sync = d.syncState(pivot.Root)
        go closeOnErr(sync)      }    } else {      results = append(append([]*fetchResult{oldPivot}, oldTail...), results...)    }    // Split around the pivot block and process the two sides via fast/full sync    if atomic.LoadInt32(&d.committed) == 0 {      latest := results[len(results)-1].Header      // If the height is above the pivot block by 2 sets, it means the pivot      // become stale in the network and it was garbage collected, move to a      // new pivot.      //      // Note, we have `reorgProtHeaderDelay` number of blocks withheld, Those      // need to be taken into account, otherwise we\\\'re detecting the pivot move      // late and will drop peers due to unavailable state!!!      if height := latest.Number.Uint64(); height >= pivot.Number.Uint64()+2*uint64(fsMinFullBlocks)-uint64(reorgProtHeaderDelay) {        log.Warn(\\\"Pivot became stale, moving\\\", \\\"old\\\", pivot.Number.Uint64(), \\\"new\\\", height-uint64(fsMinFullBlocks)+uint64(reorgProtHeaderDelay))        pivot = results[len(results)-1-fsMinFullBlocks+reorgProtHeaderDelay].Header // must exist as lower old pivot is uncommitted
        d.pivotLock.Lock()        d.pivotHeader = pivot        d.pivotLock.Unlock()
        // Write out the pivot into the database so a rollback beyond it will        // reenable fast sync        rawdb.WriteLastPivotNumber(d.stateDB, pivot.Number.Uint64())      }    }    P, beforeP, afterP := splitAroundPivot(pivot.Number.Uint64(), results)    if err := d.commitFastSyncData(beforeP, sync); err != nil {      return err    }    if P != nil {      // If new pivot block found, cancel old state retrieval and restart      if oldPivot != P {        sync.Cancel()        sync = d.syncState(P.Header.Root)
        go closeOnErr(sync)        oldPivot = P      }      // Wait for completion, occasionally checking for pivot staleness      select {      case <-sync.done:        if sync.err != nil {          return sync.err        }        if err := d.commitPivotBlock(P); err != nil {          return err        }        oldPivot = nil
      case <-time.After(time.Second):        oldTail = afterP        continue      }    }    // Fast sync done, pivot commit done, full import    if err := d.importBlockResults(afterP); err != nil {      return err    }  }}