死磕以太坊源碼分析之downloader同步

死磕以太坊源碼分析之downloader同步

需要配合注釋程式碼看：//github.com/blockchainGuide/

這篇文章篇幅較長，能看下去的是條漢子，建議收藏

希望讀者在閱讀過程中，指出問題，給個關注，一起探討。

概覽

downloader 模組的程式碼位於 eth/downloader 目錄下。主要的功能程式碼分別是：

  • downloader.go ：實現了區塊同步邏輯

  • peer.go ：對區塊各個階段的組裝，下面的各個FetchXXX 就是很依賴這個模組。

  • queue.go ：對eth/peer.go的封裝

  • statesync.go ：同步state對象

同步模式

full sync

full 模式會在資料庫中保存所有區塊數據，同步時從遠程節點同步 header 和 body 數據，而state 和 receipt 數據則是在本地計算出來的。

在 full 模式下，downloader 會同步區塊的 header 和 body 數據組成一個區塊，然後通過 blockchain 模組的 BlockChain.InsertChain 向資料庫中插入區塊。在 BlockChain.InsertChain 中，會逐個計算和驗證每個塊的 state 和 recepit 等數據，如果一切正常就將區塊數據以及自己計算得到的 state、recepit 數據一起寫入到資料庫中。

fast sync

fast 模式下，recepit 不再由本地計算，而是和區塊數據一樣，直接由 downloader 從其它節點中同步；state 數據並不會全部計算和下載，而是選一個較新的區塊（稱之為 pivot）的 state 進行下載，以這個區塊為分界，之前的區塊是沒有 state 數據的，之後的區塊會像 full 模式下一樣在本地計算 state。因此在 fast 模式下，同步的數據除了 header 和 body，還有 receipt，以及 pivot 區塊的 state。

因此 fast 模式忽略了大部分 state 數據，並且使用網路直接同步 receipt 數據的方式替換了 full 模式下的本地計算，所以比較快。

light sync

light 模式也叫做輕模式，它只對區塊頭進行同步，而不同步其它的數據。

SyncMode:

  • FullSync:從完整區塊同步整個區塊鏈歷史
  • FastSync:快速下載標題，僅在鏈頭處完全同步
  • LightSync:僅下載標題，然後終止

區塊下載流程

圖片只是大概的描述一下，實際還是要結合程式碼，所有區塊鏈相關文章合集，//github.com/blockchainGuide/

同時希望結識更多區塊鏈圈子的人，可以star上面項目，持續更新

image-20201222221031797

首先根據Synchronise開始區塊同步，通過findAncestor找到指定節點的共同祖先，並在此高度進行同步，同時開啟多個goroutine同步不同的數據：header、receipt、body。假如同步高度為 100 的區塊，必須先header同步成功同步完成才可以喚醒body和receipts的同步。

而每個部分的同步大致都是由FetchParts來完成的，裡面包含了各個Chan的配合，也會涉及不少的回調函數，總而言之多讀幾遍每次都會有不同的理解。接下來就逐步分析這些關鍵內容。

synchronise

①：確保對方的TD高於我們自己的TD

currentBlock := pm.blockchain.CurrentBlock()
	td := pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64())
	pHead, pTd := peer.Head()
	if pTd.Cmp(td) <= 0 {
		return
	}

②：開啟downloader的同步

pm.downloader.Synchronise(peer.id, pHead, pTd, mode)

進入函數：主要做了以下幾件事：

  1. d.synchronise(id, head, td, mode) ：同步過程
  2. 錯誤日誌輸出， 並刪除此peer。

進入到d.synchronise，走到最後一步d.syncWithPeer(p, hash, td)真正開啟同步。

func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode SyncMode) error {
  ...
  return d.syncWithPeer(p, hash, td)
}

syncWithPeer大概做了以下幾件事：

  1. 查找祖先findAncestor
  2. 開啟單獨goroutine分別運行以下幾個函數：
    • fetchHeaders
    • processHeaders
    • fetchbodies
    • fetchReceipts
    • processFastSyncContent
    • processFullSyncContent

接下來的文章，以及整個Downloader模組主要內容就是圍繞這幾個部分進行展開。

findAncestor

同步首要的是確定同步區塊的區間：頂部為遠程節點的最高區塊，底部為兩個節點都擁有的相同區塊的最高高度（祖先區塊）。findAncestor就是用來找祖先區塊。函數分析如下：

①：確定本地高度和遠程節點的最高高度

var (
		floor        = int64(-1) // 底部
		localHeight  uint64  // 本地最高高度
		remoteHeight = remoteHeader.Number.Uint64() // 遠程節點最高高度
	)
switch d.mode {
	case FullSync:
		localHeight = d.blockchain.CurrentBlock().NumberU64()
	case FastSync:
		localHeight = d.blockchain.CurrentFastBlock().NumberU64()
	default:
		localHeight = d.lightchain.CurrentHeader().Number.Uint64()
	}

②：計算同步的高度區間和間隔

from, count, skip, max := calculateRequestSpan(remoteHeight, localHeight)
  • from:：表示從哪個高度開始獲取區塊
  • count：表示從遠程節點獲取多少個區塊
  • skip：表示間隔，比如skip 為 2 ，獲取第一個高度為 5，則第二個就是 8
  • max：表示最大高度

③：發送獲取header的請求

go p.peer.RequestHeadersByNumber(uint64(from), count, skip, false)

④：處理上面請求接收到的header :case packet := <-d.headerCh

  1. 丟棄掉不是來自我們請求節的內容
  2. 確保返回的header數量不為空
  3. 驗證返回的headers的高度是我們所請求的
  4. 檢查是否找到共同祖先
//----①
if packet.PeerId() != p.id {
				log.Debug("Received headers from incorrect peer", "peer", packet.PeerId())
				break
			}
//-----②
headers := packet.(*headerPack).headers
			if len(headers) == 0 {
				p.log.Warn("Empty head header set")
        return 0
      }
//-----③
for i, header := range headers {
				expectNumber := from + int64(i)*int64(skip+1)
				if number := header.Number.Int64(); number != expectNumber { // 驗證這些返回的header是否是我們上面請求的headers
					p.log.Warn("Head headers broke chain ordering", "index", i, "requested", expectNumber, "received", number)
					return 0, errInvalidChain
				}
			}
//-----④
// 檢查是否找到共同祖先
			finished = true
			//注意這裡是從headers最後一個元素開始查找，也就是高度最高的區塊。
			for i := len(headers) - 1; i >= 0; i-- {
				// 跳過不在我們請求的高度區間內的區塊
				if headers[i].Number.Int64() < from || headers[i].Number.Uint64() > max {
					continue
				}
				// //檢查我們本地是否已經有某個區塊了，如果有就算是找到了共同祖先，
				//並將共同祖先的哈希和高度設置在number和hash變數中。
				h := headers[i].Hash()
				n := headers[i].Number.Uint64()

⑤：如果通過固定間隔法找到了共同祖先則返回祖先，會對其高度與 floor 變數進行驗證, floor 變數代表的是共同祖先的高度的最小值,如果找到共同祖先的高度比這個值還小，就認為是兩個節點之間分叉太大了，不再允許進行同步。如果一切正常，就返回找到的共同祖先的高度 number 變數。

if hash != (common.Hash{}) {
        if int64(number) <= floor {
            return 0, errInvalidAncestor
        }
        return number, nil
    }

⑥：如果固定間隔法沒有找到祖先則通過二分法來查找祖先，這部分可以思想跟二分法演算法類似，有興趣的可以細看。

queue詳解

queue對象和Downloader對象是相互作用的，Downloader的很多功能離不開他，接下來我們介紹一下這部分內容，但是本節，可以先行跳過，等到了閱讀下面的關於Queue調用的一些函數部分再回過來閱讀這部分講解。

queue結構體

type queue struct {
  mode SyncMode // 同步模式
  
  // header處理相關
  headerHead      common.Hash   //最後一個排隊的標頭的哈希值以驗證順序
  headerTaskPool  map[uint64]*types.Header  //待處理的標頭檢索任務，將起始索引映射到框架標頭
  headerTaskQueue *prque.Prque  //骨架索引的優先順序隊列，以獲取用於的填充標頭
  headerPeerMiss map[string]map[uint64]struct{} //已知不可用的對等頭批處理集
  headerPendPool map[string]*fetchRequest //當前掛起的頭檢索操作
  headerResults []*types.Header //結果快取累積完成的頭
  headerProced int //從結果中拿出來已經處理的header
  headerContCh chan bool //header下載完成時通知的頻道
  
  blockTaskPool  map[common.Hash]*types.Header //待處理的塊（body）檢索任務，將哈希映射到header
  blockTaskQueue *prque.Prque //標頭的優先順序隊列,以用於獲取塊（bodies）
  blockPendPool map[string]*fetchRequest //當前的正在處理的塊（body)檢索操作
  blockDonePool map[common.Hash]struct{} //已經完成的塊（body)
  
	receiptTaskPool map[common.Hash]*types.Header //待處理的收據檢索任務，將哈希映射到header
	receiptTaskQueue *prque.Prque //標頭的優先順序隊列,以用於獲取收據
	receiptPendPool map[string]*fetchRequest //當前的正在處理的收據檢索操作
	receiptDonePool map[common.Hash]struct{} //已經完成的收據
	
	resultCache []*fetchResult //下載但尚未交付獲取結果
	resultOffset uint64 //區塊鏈中第一個快取的獲取結果的偏移量
	resultSize common.StorageSize // 塊的近似大小

	lock   *sync.Mutex
	active *sync.Cond
	closed bool
  
}

主要細分功能

數據下載開始安排任務

  • ScheduleSkeleton:將一批header檢索任務添加到隊列中，以填充已檢索的header skeleton
  • Schedule:用來準備對一些 body 和 receipt 數據的下載

數據下載中的各類狀態

  • pending

    pending表示待檢索的XXX請求的數量，包括了：PendingHeaders、PendingBlocks、PendingReceipts，分別都是對應取XXXTaskQueue的長度。

  • InFlight

    InFlight表示是否有正在獲取XXX的請求，包括：InFlightHeaders、InFlightBlocks、InFlightReceipts，都是通過判斷len(q.receiptPendPool) > 0 來確認。

  • ShouldThrottle

    ShouldThrottle表示檢查是否應該限制下載XXX，包括:ShouldThrottleBlocks、ShouldThrottleReceipts，主要是為了防止下載過程中本地記憶體佔用過大。

  • Reserve

    Reserve通過構造一個 fetchRequest 結構並返回，向調用者提供指定數量的待下載的數據的資訊（queue 內部會將這些數據標記為「正在下載」）。調用者使用返回的 fetchRequest 數據向遠程節點發起新的獲取數據的請求。包括：ReserveHeaders、ReserveBodies、ReserveReceipts。

  • Cancel

    Cance用來撤消對 fetchRequest 結構中的數據的下載（queue 內部會將這些數據重新從「正在下載」的狀態更改為「等待下載」）。包括：CancelHeaders、CancelBodies、CancelReceipts。

  • expire

    expire檢查正在執行中的請求是否超過了超時限制，包括：ExpireHeaders、ExpireBodies、ExpireReceipts。

  • Deliver

    當有數據下載成功時，調用者會使用 deliver 功能用來通知 queue 對象。包括：DeliverHeaders、DeliverBodies、DeliverReceipts。

數據下載完成獲取區塊數據

  • RetrieveHeaders
    在填充 skeleton 完成後，queue.RetrieveHeaders 用來獲取整個 skeleton 中的所有 header。
  • Results
    queue.Results 用來獲取當前的 header、body 和 receipt（只在 fast 模式下） 都已下載成功的區塊（並將這些區塊從 queue 內部移除）

函數實現

ScheduleSkeleton

queue.ScheduleSkeleton主要是為了填充skeleton，它的參數是要下載區塊的起始高度和所有 skeleton 區塊頭，最核心的內容則是下面這段循環：

func (q *queue) ScheduleSkeleton(from uint64, skeleton []*types.Header) {
    ......
    for i, header := range skeleton {
        index := from + uint64(i*y)
        q.headerTaskPool[index] = header
        q.headerTaskQueue.Push(index, -int64(index))
    }
}

假設已確定需要下載的區塊高度區間是從 10 到 46，MaxHeaderFetch 的值為 10，那麼這個高度區塊就會被分成 3 組：10 – 19，20 – 29，30 – 39，而 skeleton 則分別由高度為 19、29、39 的區塊頭組成。循環中的 index 變數實際上是每一組區塊中的第一個區塊的高度（比如 10、20、30），queue.headerTaskPool 實際上是一個每一組區塊中第一個區塊的高度到最後一個區塊的 header 的映射

headerTaskPool = {
  10: headerOf_19,
	20: headerOf_20,
	30: headerOf_39,
}

ReserveHeaders

reserve 用來獲取可下載的數據。

reserve  = func(p *peerConnection, count int) (*fetchRequest, bool, error) {
			return d.queue.ReserveHeaders(p, count), false, nil
		}

func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest {
  if _, ok := q.headerPendPool[p.id]; ok {
		return nil
	} //①
  ...
  send, skip := uint64(0), []uint64{}
	for send == 0 && !q.headerTaskQueue.Empty() {
		from, _ := q.headerTaskQueue.Pop()
		if q.headerPeerMiss[p.id] != nil {
			if _, ok := q.headerPeerMiss[p.id][from.(uint64)]; ok {
				skip = append(skip, from.(uint64))
				continue
			}
		}
		send = from.(uint64) // ②
	}
  
 ...
  for _, from := range skip {
		q.headerTaskQueue.Push(from, -int64(from))
	} // ③
  ...
  request := &fetchRequest{
		Peer: p,
		From: send,
		Time: time.Now(),
	}
	q.headerPendPool[p.id] = request // ④
  
}

①：根據headerPendPool來判斷遠程節點是否正在下載數據資訊。

②：從headerTaskQueue取出值作為本次請求的起始高度，賦值給send變數，在這個過程中會排除headerPeerMiss所記錄的節點下載數據失敗的資訊。

③：將失敗的任務再重新寫回task queue

④：利用send變數構造fetchRequest結構，此結構是用來作為FetchHeaders來使用的：

fetch = func(p *peerConnection, req *fetchRequest) error { 
    return p.FetchHeaders(req.From, MaxHeaderFetch) 
}

至此，ReserveHeaders會從任務隊列里選擇最小的起始高度並構造fetchRequest傳遞給fetch獲取數據。

DeliverHeaders

deliver = func(packet dataPack) (int, error) {
			pack := packet.(*headerPack)
			return d.queue.DeliverHeaders(pack.peerID, pack.headers, d.headerProcCh)
		}

①：如果發現下載數據的節點沒有在 queue.headerPendPool 中，就直接返回錯誤；否則就繼續處理，並將節點記錄從 queue.headerPendPool 中刪除。

request := q.headerPendPool[id]
	if request == nil {
		return 0, errNoFetchesPending
	}
	headerReqTimer.UpdateSince(request.Time)
	delete(q.headerPendPool, id)

②：驗證headers

包括三方面驗證：

  1. 檢查起始區塊的高度和哈希
  2. 檢查高度的連接性
  3. 檢查哈希的連接性
if accepted {
		//檢查起始區塊的高度和哈希
		if headers[0].Number.Uint64() != request.From {
			...
			accepted = false
		} else if headers[len(headers)-1].Hash() != target {
			...
			accepted = false
		}
	}
	if accepted {
		for i, header := range headers[1:] {
			hash := header.Hash() // 檢查高度的連接性
			if want := request.From + 1 + uint64(i); header.Number.Uint64() != want {
				...
			}
			if headers[i].Hash() != header.ParentHash { // 檢查哈希的連接性
				...
			}
		}
	}

③： 將無效數據存入headerPeerMiss，並將這組區塊起始高度重新放入headerTaskQueue

if !accepted {
	...
		miss := q.headerPeerMiss[id]
		if miss == nil {
			q.headerPeerMiss[id] = make(map[uint64]struct{})
			miss = q.headerPeerMiss[id]
		}
		miss[request.From] = struct{}{}
		q.headerTaskQueue.Push(request.From, -int64(request.From))
		return 0, errors.New("delivery not accepted")
	}

④：保存數據，並通知headerProcCh處理新的header

if ready > 0 {
		process := make([]*types.Header, ready)
		copy(process, q.headerResults[q.headerProced:q.headerProced+ready])
		select {
		case headerProcCh <- process:
			q.headerProced += len(process)
		default:
		}
	}

⑤：發送消息給.headerContCh，通知skeleton 都被下載完了

if len(q.headerTaskPool) == 0 {
		q.headerContCh <- false
	}

DeliverHeaders 會對數據進行檢驗和保存，並發送 channel 消息給 Downloader.processHeaders 和 Downloader.fetchParts的 wakeCh 參數。

Schedule

processHeaders在處理header數據的時候，會調用queue.Schedule 為下載 body 和 receipt 作準備。

inserts := d.queue.Schedule(chunk, origin)

func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
	inserts := make([]*types.Header, 0, len(headers))
	for _, header := range headers {
    //校驗
    ...
		q.blockTaskPool[hash] = header
		q.blockTaskQueue.Push(header, -int64(header.Number.Uint64()))

		if q.mode == FastSync {
			q.receiptTaskPool[hash] = header
			q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64()))
		}
		inserts = append(inserts, header)
		q.headerHead = hash
		from++
	}
	return inserts
}

這個函數主要就是將資訊寫入到body和receipt隊列，等待調度。

ReserveBody&Receipt

在 queue 中準備好了 body 和 receipt 相關的數據， processHeaders最後一段，是喚醒下載Bodyies和Receipts的關鍵程式碼，會通知 fetchBodies 和 fetchReceipts 可以對各自的數據進行下載了。

for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {
				select {
				case ch <- true:
				default:
				}
			}

而fetchXXX 會調用fetchParts，邏輯類似上面的的，reserve最終則會調用reserveHeaders，deliver 最終調用的是 queue.deliver.

先來分析reserveHeaders：

①：如果沒有可處理的任務，直接返回

if taskQueue.Empty() {
        return nil, false, nil
    }

②：如果參數給定的節點正在下載數據，返回

 if _, ok := pendPool[p.id]; ok {
        return nil, false, nil
    }

③：計算 queue 對象中的快取空間還可以容納多少條數據

space := q.resultSlots(pendPool, donePool)

④：從 「task queue」 中依次取出任務進行處理

主要實現以下功能：

  • 計算當前 header 在 queue.resultCache 中的位置，然後填充 queue.resultCache 中相應位置的元素
  • 處理空區塊的情況，若為空不下載。
  • 處理遠程節點缺少這個當前區塊數據的情況，如果發現這個節點曾經下載當前數據失敗過，就不再讓它下載了。

注意：resultCache 欄位用來記錄所有正在被處理的數據的處理結果，它的元素類型是 fetchResult 。它的 Pending 欄位代表當前區塊還有幾類數據需要下載。這裡需要下載的數據最多有兩類：body 和 receipt，full 模式下只需要下載 body 數據，而 fast 模式要多下載一個 receipt 數據。

for proc := 0; proc < space && len(send) < count && !taskQueue.Empty(); proc++ {
		header := taskQueue.PopItem().(*types.Header)
		hash := header.Hash()
		index := int(header.Number.Int64() - int64(q.resultOffset))
		if index >= len(q.resultCache) || index < 0 {
			....
		}
		if q.resultCache[index] == nil {
			components := 1
			if q.mode == FastSync {
				components = 2
			}
			q.resultCache[index] = &fetchResult{
				Pending: components,
				Hash:    hash,
				Header:  header,
			}
		}
  
		if isNoop(header) {
			donePool[hash] = struct{}{}
			delete(taskPool, hash)

			space, proc = space-1, proc-1
			q.resultCache[index].Pending--
			progress = true
			continue
		}
		if p.Lacks(hash) {
			skip = append(skip, header)
		} else {
			send = append(send, header)
		}
	}

最後就是構造 fetchRequest 結構並返回。

DeliverBodies&Receipts

body 或 receipt 數據都已經通過 reserve 操作構造了 fetchRequest 結構並傳給 fetch，接下來就是等待數據的到達,數據下載成功後，會調用 queue 對象的 deliver 方法進行傳遞，包括 queue.DeliverBodies 和 queue.DeliverReceipts。這兩個方法都以不同的參數調用了 queue.deliver 方法:

①：如果下載的數據數量為 0，則把所有此節點此次下載的數據標記為「缺失」

if results == 0 {
		for _, header := range request.Headers {
			request.Peer.MarkLacking(header.Hash())
		}
	}

②：循環處理數據，通過調用reconstruct 填充 resultCache[index] 中的相應的欄位

for i, header := range request.Headers {
  ...
  if err := reconstruct(header, i, q.resultCache[index]); err != nil {
			failure = err
			break
		}
}

③：驗證resultCache 中的數據，其對應的 request.Headers 中的 header 都應為 nil，若不是則說明驗證未通過，需要假如到task queue重新下載

for _, header := range request.Headers {
		if header != nil {
			taskQueue.Push(header, -int64(header.Number.Uint64()))
		}
	}

④：如果有數據被驗證通過且寫入 queue.resultCache 中了（accepted > 0），發送 queue.active 消息。Results 會等待這這個訊號。

Results

當(header、body、receipt)都下載完，就要將區塊寫入到資料庫了，queue.Results 就是用來返回所有目前已經下載完成的數據，它在 Downloader.processFullSyncContent 和 Downloader.processFastSyncContent 中被調用。程式碼比較簡單就不多說了。

到此為止queue對象就分析的差不多了。

同步headers

fetchHeaders

同步headers 是是由函數fetchHeaders來完成的。

fetchHeaders的大致思想：

同步header的數據會被填充到skeleton，每次從遠程節點獲取區塊數據最大為MaxHeaderFetch（192），所以要獲取的區塊數據如果大於192 ，會被分成組，每組MaxHeaderFetch，剩餘的不足192個的不會填充進skeleton，具體步驟如下圖所示：

image-20201219111103965

此種方式可以避免從同一節點下載過多錯誤數據，如果我們連接到了一個惡意節點，它可以創造一個鏈條很長且TD值也非常高的區塊鏈數據。如果我們的區塊從 0 開始全部從它那同步，也就下載了一些根本不被別人承認的數據。如果我只從它那同步 MaxHeaderFetch 個區塊，然後發現這些區塊無法正確填充我之前的 skeleton（可能是 skeleton 的數據錯了，或者用來填充 skeleton 的數據錯了），就會丟掉這些數據。

接下來查看下程式碼如何實現：

①：發起獲取header的請求

如果是下載skeleton，則會從高度 from+MaxHeaderFetch-1 開始（包括），每隔 MaxHeaderFetch-1 的高度請求一個 header，最多請求 MaxSkeletonSize 個。如果不是的話，則要獲取完整的headers 。

②：等待並處理headerCh中的header數據

2.1 確保遠程節點正在返回我們需要填充skeleton所需的header

if packet.PeerId() != p.id {
				log.Debug("Received skeleton from incorrect peer", "peer", packet.PeerId())
				break
			}

2.2 如果skeleton已經下載完畢，則需要繼續填充skeleton

if packet.Items() == 0 && skeleton {
				skeleton = false
				getHeaders(from)
				continue
			}

2.3 整個skeleton填充完成，並且沒有要獲取的header了，要通知headerProcCh全部完成

if packet.Items() == 0 {
				//下載pivot時不要中止標頭的提取
				if atomic.LoadInt32(&d.committed) == 0 && pivot <= from {
					p.log.Debug("No headers, waiting for pivot commit")
					select {
					case <-time.After(fsHeaderContCheck):
						getHeaders(from)
						continue
					case <-d.cancelCh:
						return errCanceled
					}
				}
				//完成Pivot操作（或不進行快速同步），並且沒有頭文件，終止該過程
				p.log.Debug("No more headers available")
				select {
				case d.headerProcCh <- nil:
					return nil
				case <-d.cancelCh:
					return errCanceled
				}
			}

2.4 當header有數據並且是在獲取skeleton的時候，調用fillHeaderSkeleton填充skeleton

if skeleton {
				filled, proced, err := d.fillHeaderSkeleton(from, headers)
				if err != nil {
					p.log.Debug("Skeleton chain invalid", "err", err)
					return errInvalidChain
				}
				headers = filled[proced:]
				from += uint64(proced)
			}

2.5 如果當前處理的不是 skeleton，表明區塊同步得差不多了，處理尾部的一些區塊

判斷本地的主鏈高度與新收到的 header 的最高高度的高度差是否在 reorgProtThreshold 以內，如果不是，就將高度最高的 reorgProtHeaderDelay 個 header 丟掉。

if head+uint64(reorgProtThreshold) < headers[n-1].Number.Uint64() {
						delay := reorgProtHeaderDelay
						if delay > n {
							delay = n
						}
						headers = headers[:n-delay]
					}

2.6 如果還有 header 未處理，發給 headerProcCh 進行處理，Downloader.processHeaders 會等待這個 channel 的消息並進行處理；

if len(headers) > 0 {
				...
				select {
				case d.headerProcCh <- headers:
				case <-d.cancelCh:
					return errCanceled
				}
				from += uint64(len(headers))
  getHeaders(from)
}

2.7 如果沒有發送標頭，或者所有標頭等待 fsHeaderContCheck 秒，再次調用 getHeaders 請求區塊

p.log.Trace("All headers delayed, waiting")
				select {
				case <-time.After(fsHeaderContCheck):
					getHeaders(from)
					continue
				case <-d.cancelCh:
					return errCanceled
				}

這段程式碼後來才加上的，其 commit 的記錄在這裡，而 「pull request」 在這裡。從 「pull request」 中作者的解釋我們可以了解這段程式碼的邏輯和功能：這個修改主要是為了解決經常出現的 「invalid hash chain」 錯誤，出現這個錯誤的原因是因為在我們上一次從遠程節點獲取到一些區塊並將它們加入到本地的主鏈的過程中，遠程節點發生了 reorg 操作（參見這篇文章里關於「主鏈與側鏈」的介紹 ）；當我們再次根據高度請求新的區塊時，對方返回給我們的是它的新的主鏈上的區塊，而我們沒有這個鏈上的歷史區塊，因此在本地寫入區塊時就會返回 「invalid hash chain」 錯誤。

要想發生 「reorg」 操作，就需要有新區塊加入。在以太坊主網上，新產生一個區塊的間隔是 10 秒到 20 秒左右。一般情況下，如果僅僅是區塊數據，它的同步速度還是很快的，每次下載也有最大數量的限制。所以在新產生一個區塊的這段時間裡，足夠同步完成一組區塊數據而對方節點不會發生 「reorg」 操作。但是注意剛才說的「僅僅是區塊數據」的同步較快，state 數據的同步就非常慢了。簡單來說在完成同步之前可能會有多個 「pivot」 區塊，這些區塊的 state 數據會從網路上下載，這就大大拖慢了整個區塊的同步速度，使得本地在同步一組區塊的同時對方發生 「reorg」 操作的機率大大增加。

作者認為這種情況下發生的 「reorg」 操作是由新產生的區塊的競爭引起的，所以最新的幾個區塊是「不穩定的」，如果本次同步的區塊數量較多（也就是我們同步時消耗的時間比較長）（在這裡「本次同步的區數數量較多」的表現是新收到的區塊的最高高度與本地資料庫中的最高高度的差距大於 reorgProtThreshold），那麼在同步時可以先避免同步最新區塊，這就是 reorgProtThreshold 和 reorgProtHeaderDelay 這個變數的由來。

至此，Downloader.fetchHeaders 方法就結束了，所有的區塊頭也就同步完成了。在上面我們提到填充skeleton的時候，是由fillHeaderSkeleton函數來完成，接下來就要細講填充skeleton的細節。

fillHeaderSkeleton

首先我們知道以太坊在同步區塊時，先確定要下載的區塊的高度區間，然後將這個區間按 MaxHeaderFetch 切分成很多組，每一組的最後一個區塊組成了 「skeleton」（最後一組不滿 MaxHeaderFetch 個區塊不算作一組）。不清楚的可以查看上面的圖。

①：將一批header檢索任務添加到隊列中，以填充skeleton。

這個函數參照上面queue詳解的分析

func (q queue) ScheduleSkeleton(from uint64, skeleton []types.Header) {}

②：調用fetchParts 獲取headers數據

fetchParts是很核心的函數，下面的Fetchbodies和FetchReceipts都會調用。先來大致看一下fetchParts的結構：

func (d *Downloader) fetchParts(...) error {
  ...
  for {
		select {
		case <-d.cancelCh:
		case packet := <-deliveryCh:
		case cont := <-wakeCh:
		case <-ticker.C:
		case <-update:
		...
	}
}

簡化下來就是這 5 個channel在處理，前面 4 個channel負責循環等待消息，update用來等待其他 4 個channel的通知來處理邏輯，先分開分析一個個的channel。

2.1 deliveryCh 傳遞下載的數據

deliveryCh 作用就是傳遞下載的數據，當有數據被真正下載下來時，就會給這個 channel 發消息將數據傳遞過來。這個 channel 對應的分別是：d.headerCh、d.bodyCh、d.receiptCh，而這三個 channel 分別在以下三個方法中被寫入數據：DeliverHeaders、DeliverBodies、DeliverReceipts。 看下deliveryCh如何處理數據：

case packet := <-deliveryCh:
			if peer := d.peers.Peer(packet.PeerId()); peer != nil {
				accepted, err := deliver(packet)//傳遞接收到的數據塊並檢查鏈有效性
				if err == errInvalidChain {
					return err
        }
				if err != errStaleDelivery {
					setIdle(peer, accepted)
				}
				switch {
				case err == nil && packet.Items() == 0:
					...
				case err == nil:
				...
				}
			}
			select {
			case update <- struct{}{}:
			default:
			}

收到下載數據後判斷節點是否有效，如果節點沒有被移除，則會通過deliver傳遞接收到的下載數據。如果沒有任何錯誤，則通知update處理。

要注意deliver是一個回調函數，它調用了 queue 對象的 Deliver 方法：queue.DeliverHeaders、queue.DeliverBodies、queue.DeliverReceipts，在收到下載數據就會調用此回調函數（queue相關函數分析參照queue詳解部分）。

在上面處理錯誤部分，有一個setIdle函數，它也是回調函數，其實現都是調用了 peerConnection 對象的相關方法：SetHeadersIdle、SetBodiesIdle、SetReceiptsIdle。它這個函數是指某些節點針對某類數據是空閑的，比如header、bodies、receipts，如果需要下載這幾類數據，就可以從空閑的節點下載這些數據。

2.2 wakeCh 喚醒fetchParts ，下載新數據或下載已完成

case cont := <-wakeCh:
			if !cont {
				finished = true
			}
			select {
			case update <- struct{}{}:
			default:
			}

首先我們通過調用fetchParts傳遞的參數知道，wakeCh 的值其實是 queue.headerContCh。在 queue.DeliverHeaders 中發現所有需要下戴的 header 都下載完成了時，才會發送 false 給這個 channel。fetchParts 在收到這個消息時，就知道沒有 header 需要下載了。程式碼如下：

func (q *queue) DeliverHeaders(......) (int, error) {
    ......
    if len(q.headerTaskPool) == 0 {
        q.headerContCh <- false
    }
    ......
}

同樣如此，body和receipt則是bodyWakeCh和receiptWakeCh，在 processHeaders 中，如果所有 header 已經下載完成了，那麼發送 false 給這兩個 channel，通知它們沒有新的 header 了。 body 和 receipt 的下載依賴於 header,需要 header 先下載完成才能下載，所以對於下戴 body 或 receipt 的 fetchParts 來說，收到這個 wakeCh 就代表不會再有通知讓自己下載數據了.

func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) error {
    for {
        select {
        case headers := <-d.headerProcCh:
            if len(headers) == 0 {
                for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {
                    select {
                    case ch <- false:
                    case <-d.cancelCh:
                    }
                }
						...
            }
            ...
            for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {
                select {
                case ch <- true:
                default:
                }
            }
        }
    }
}

2.3 ticker 負責周期性的激活 update進行消息處理

case <-ticker.C:
			select {
			case update <- struct{}{}:
			default:
			}

2.4 update （處理此前幾個channel的數據）(重要)

2.4.1 判斷是否有效節點，並獲取超時數據的資訊

獲取超時數據的節點ID和數據數量，如果大於兩個的話，就將這個節點設置為空閑狀態(setIdle)，小於兩個的話直接斷開節點連接。

expire 是一個回調函數，會返回當前所有的超時數據資訊。這個函數的實際實現都是調用了 queue 對象的 Expire 方法：ExpireHeaders、ExpireBodies、ExpireReceipts,此函數會統計當前正在下載的數據中，起始時間與當前時間的差距超過給定閾值（downloader.requestTTL 方法的返回值）的數據，並將其返回。

if d.peers.Len() == 0 {
				return errNoPeers
			}
for pid, fails := range expire() {
  if peer := d.peers.Peer(pid); peer != nil {
    if fails > 2 {
						...
						setIdle(peer, 0)
					} else {
					...
						if d.dropPeer == nil {
						} else {
							d.dropPeer(pid)
							....
						}
					}
  }

2.4.2 處理完超時數據，判斷是否還有下載的數據

如果沒有其他可下載的內容，請等待或終止，這裡pending()和inFlight()都是回調函數，pending分別對應了queue.PendingHeaders、queue.PendingBlocks、queue.PendingReceipts,用來返回各自要下載的任務數量。inFlight()分別對應了queue.InFlightHeaders、queue.InFlightBlocks、queue.InFlightReceipts,用來返回正在下載的數據數量。

if pending() == 0 {
				if !inFlight() && finished {
				...
					return nil
				}
				break
			}

2.4.3 使用空閑節點，調用fetch函數發送數據請求

Idle()回調函數在上面已經提過了，throttle()回調函數則分別對queue.ShouldThrottleBlocks、queue.ShouldThrottleReceipts,用來表示是否應該下載bodies或者receipts。

reserve函數分別對應queue.ReserveHeaders、queue.ReserveBodies、queue.ReserveReceipts,用來從從下載任務中選取一些可以下載的任務，並構造一個 fetchRequest 結構。它還返回一個 process 變數，標記著是否有空的數據正在被處理。比如有可能某區塊中未包含任何一條交易，因此它的 body 和 receipt 都是空的，這種數據其實是不需要下載的。在 queue 對象的 Reserve 方法中，會對這種情況進行識別。如果遇到空的數據，這些數據會被直接標記為下載成功。在方法返回時，就將是否發生過「直接標記為下載成功」的情況返回。

capacity回調函數分別對應peerConnection.HeaderCapacity、peerConnection.BlockCapacity、peerConnection.ReceiptCapacity,用來決定下載需要請求數據的個數。

fetch回調函數分別對應peer.FetchHeaders、peer.Fetchbodies、peer.FetchReceipts,用來發送獲取各類數據的請求。

progressed, throttled, running := false, false, inFlight()
			idles, total := idle()
			for _, peer := range idles {
				if throttle() {
					...
        }
				if pending() == 0 {
					break
				}
				request, progress, err := reserve(peer, capacity(peer))
				if err != nil {
					return err
				}
				if progress {
					progressed = true
				}
        if request == nil {
					continue
				}
				if request.From > 0 {
				...
				}
				...
				if err := fetch(peer, request); err != nil {
				...
			}
			if !progressed && !throttled && !running && len(idles) == total && pending() > 0 {
				return errPeersUnavailable
			}

簡單來概括這段程式碼就是：使用空閑節點下載數據，判斷是否需要暫停，或者數據是否已經下載完成；之後選取數據進行下載；最後，如果沒有遇到空塊需要下載、且沒有暫停下載和所有有效節點都空閑和確實有數據需要下載，但下載沒有運行起來，就返回 errPeersUnavailable 錯誤。

到此為止fetchParts函數就分析的差不多了。裡面涉及的跟queue.go相關的一些函數都在queue詳解小節里介紹了。

processHeaders

通過headerProcCh接收header數據，並處理的過程是在processHeaders函數中完成的。整個處理過程集中在：case headers := <-d.headerProcCh中:

①：如果headers的長度為0 ，則會有以下操作：

1.1 通知所有人header已經處理完畢

for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {
					select {
					case ch <- false:
					case <-d.cancelCh:
					}
				}

1.2 若沒有檢索到任何header，說明他們的TD小於我們的，或者已經通過我們的fetcher模組進行了同步。

if d.mode != LightSync {
					head := d.blockchain.CurrentBlock()
					if !gotHeaders && td.Cmp(d.blockchain.GetTd(head.Hash(), head.NumberU64())) > 0 {
						return errStallingPeer
					}
				}

1.3 如果是fast或者light 同步，確保傳遞了header

if d.mode == FastSync || d.mode == LightSync {
					head := d.lightchain.CurrentHeader()
					if td.Cmp(d.lightchain.GetTd(head.Hash(), head.Number.Uint64())) > 0 {
						return errStallingPeer
					}
				}

②：如果headers的長度大於 0

2.1 如果是fast或者light 同步，調用ightchain.InsertHeaderChain()寫入header到leveldb資料庫；

if d.mode == FastSync || d.mode == LightSync {
  ....
  d.lightchain.InsertHeaderChain(chunk, frequency);
  ....
}

2.2 如果是fast或者full sync模式，則調用 d.queue.Schedule進行內容(body和receipt)檢索。

if d.mode == FullSync || d.mode == FastSync {
  ...
  inserts := d.queue.Schedule(chunk, origin)
  ...
}

③：如果找到更新的塊號，則要發訊號通知新任務

if d.syncStatsChainHeight < origin {
				d.syncStatsChainHeight = origin - 1
			}
for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {
				select {
				case ch <- true:
				default:
				}
			}

到此處理Headers的分析就完成了。

同步bodies

同步bodies 則是由fetchBodies函數完成的。

fetchBodies

同步bodies的過程跟同步header類似，大致講下步驟：

  1. 調用fetchParts
  2. ReserveBodies()從bodyTaskPool中取出要同步的body；
  3. 調用fetch，也就是調用這裡的FetchBodies從節點獲取body，發送GetBlockBodiesMsg消息；
  4. 收到bodyCh的數據後，調用deliver函數，將Transactions和Uncles寫入resultCache。

同步Receipts

fetchReceipts

同步receipts的過程跟同步header類似，大致講下步驟：

  1. 調用fetchParts()
  2. ReserveBodies()從ReceiptTaskPool中取出要同步的Receipt
  3. 調用這裡的FetchReceipts從節點獲取receipts，發送GetReceiptsMsg消息；
  4. 收到receiptCh的數據後，調用deliver函數，將Receipts寫入resultCache。

同步狀態

這裡我們講兩種模式下的狀態同步：

  • fullSync: processFullSyncContent，full模式下Receipts沒有快取到resultCache中，直接先從快取中取出body數據，然後執行交易生成狀態，最後寫入區塊鏈。
  • fastSync:processFastSyncContent：fast模式的Receipts、Transaction、Uncles都在resultCache中，所以還需要下載”state”，進行校驗，再寫入區塊鏈。

接下來大致的討論下這兩種方式。

processFullSyncContent

func (d *Downloader) processFullSyncContent() error {
	for {
		results := d.queue.Results(true)
		...
		if err := d.importBlockResults(results); err != nil ...
	}
}

func (d *Downloader) importBlockResults(results []*fetchResult) error {
	...
	select {
...
	blocks := make([]*types.Block, len(results))
	for i, result := range results {
		blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles)
	}
	if index, err := d.blockchain.InsertChain(blocks); err != nil {
		....
}

直接從result中獲取數據並生成block，直接插入區塊鏈中，就結束了。

processFastSyncContent

fast模式同步狀態內容比較多，大致也就如下幾部分，我們開始簡單分析以下。

①：下載最新的區塊狀態

sync := d.syncState(latest.Root)

我們直接用一張圖來表示整個大致流程：

image-20201223100153241

具體的程式碼讀者自己翻閱，大致就是這麼個簡單過程。

②：計算出pivot塊

pivot為latestHeight - 64，調用splitAroundPivot()方法以pivot為中心，將results分為三個部分：beforeP，P，afterP；

pivot := uint64(0)
	if height := latest.Number.Uint64(); height > uint64(fsMinFullBlocks) {
		pivot = height - uint64(fsMinFullBlocks)
	}

P, beforeP, afterP := splitAroundPivot(pivot, results)

③： 對beforeP的部分調用commitFastSyncData，將body和receipt都寫入區塊鏈

d.commitFastSyncData(beforeP, sync);

④：對P的部分更新狀態資訊為P block的狀態，把P對應的result（包含body和receipt）調用commitPivotBlock插入本地區塊鏈中，並調用FastSyncCommitHead記錄這個pivot的hash值，存在downloader中，標記為快速同步的最後一個區塊hash值；

if err := d.commitPivotBlock(P); err != nil {
					return err
				}

⑤：對afterP調用d.importBlockResults，將body插入區塊鏈，而不插入receipt。因為是最後 64 個區塊，所以此時資料庫中只有header和body，沒有receipt和狀態，要通過fullSync模式進行最後的同步。

if err := d.importBlockResults(afterP); err != nil {
			return err
		}

到此為止整個Downloader同步完成了。

參考

//mindcarver.cn

//github.com/ethereum/go-ethereum/pull/1889

//yangzhe.me/2019/05/09/ethereum-downloader/#fetchparts

Tags:

VirMach 便宜 VPS

QNews

QNews