queue給downloader提供了排程功能和限流的功能。 通過呼叫Schedule/ScheduleSkeleton來申請對任務進行排程，然後呼叫ReserveXXX方法來領取排程完成的任務，並在downloader裡面的執行緒來執行，呼叫DeliverXXX方法把下載完的資料給queue。 最後通過WaitResults來獲取已經完成的任務。中間還有一些對任務的額外控制，ExpireXXX用來控制任務是否超時， CancelXXX用來取消任務。

Schedule方法

Schedule呼叫申請對一些區塊頭進行下載排程。可以看到做了一些合法性檢查之後，把任務插入了blockTaskPool，receiptTaskPool，receiptTaskQueue，receiptTaskPool。
TaskPool是Map，用來記錄header的hash是否存在。 TaskQueue是優先順序佇列，優先順序是區塊的高度的負數， 這樣區塊高度越小的優先順序越高，就實現了首先排程小的任務的功能。

// Schedule adds a set of headers for the download queue for scheduling, returning
// the new headers encountered.
// from表示headers裡面第一個元素的區塊高度。 返回值返回了所有被接收的header
func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
    q.lock.Lock()
    defer q.lock.Unlock()

    // Insert all the headers prioritised by the contained block number
    inserts := make([]*types.Header, 0, len(headers))
    for _, header := range headers {
        // Make sure chain order is honoured and preserved throughout
        hash := header.Hash()
        if header.Number == nil || header.Number.Uint64() != from {
            log.Warn("Header broke chain ordering", "number", header.Number, "hash", hash, "expected", from)
            break
        }
        //headerHead儲存了最後一個插入的區塊頭， 檢查當前區塊是否正確的連結。
        if q.headerHead != (common.Hash{}) && q.headerHead != header.ParentHash {
            log.Warn("Header broke chain ancestry", "number", header.Number, "hash", hash)
            break
        }
        // Make sure no duplicate requests are executed
        // 檢查重複，這裡直接continue了，那不是from對不上了。
        if _, ok := q.blockTaskPool[hash]; ok {
            log.Warn("Header  already scheduled for block fetch", "number", header.Number, "hash", hash)
            continue
        }
        if _, ok := q.receiptTaskPool[hash]; ok {
            log.Warn("Header already scheduled for receipt fetch", "number", header.Number, "hash", hash)
            continue
        }
        // Queue the header for content retrieval
        q.blockTaskPool[hash] = header
        q.blockTaskQueue.Push(header, -float32(header.Number.Uint64()))

        if q.mode == FastSync && header.Number.Uint64() <= q.fastSyncPivot {
            // Fast phase of the fast sync, retrieve receipts too
            // 如果是快速同步模式，而且區塊高度也小於pivot point. 那麼還要獲取receipt
            q.receiptTaskPool[hash] = header
            q.receiptTaskQueue.Push(header, -float32(header.Number.Uint64()))
        }
        inserts = append(inserts, header)
        q.headerHead = hash
        from++
    }
    return inserts
}
 

ReserveXXX

ReserveXXX方法用來從queue裡面領取一些任務來執行。downloader裡面的goroutine會呼叫這個方法來領取一些任務來執行。 這個方法直接呼叫了reserveHeaders方法。 所有的ReserveXXX方法都會呼叫reserveHeaders方法，除了傳入的引數有一些區別。

// ReserveBodies reserves a set of body fetches for the given peer, skipping any
// previously failed downloads. Beside the next batch of needed fetches, it also
// returns a flag whether empty blocks were queued requiring processing.
func (q *queue) ReserveBodies(p *peerConnection, count int) (*fetchRequest, bool, error) {
    isNoop := func(header *types.Header) bool {
        return header.TxHash == types.EmptyRootHash && header.UncleHash == types.EmptyUncleHash
    }
    q.lock.Lock()
    defer q.lock.Unlock()

    return q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, isNoop)
}
 

reserveHeaders

// reserveHeaders reserves a set of data download operations for a given peer,
// skipping any previously failed ones. This method is a generic version used
// by the individual special reservation functions.
// reserveHeaders為指定的peer保留一些下載操作，跳過之前的任意錯誤。 這個方法單獨被指定的保留方法呼叫。
// Note, this method expects the queue lock to be already held for writing. The
// reason the lock is not obtained in here is because the parameters already need
// to access the queue, so they already need a lock anyway.
// 這個方法呼叫的時候，假設已經獲取到鎖，這個方法裡面沒有鎖的原因是引數已經傳入到函式裡面了，所以呼叫的時候就需要獲取鎖。
func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
    pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, isNoop func(*types.Header) bool) (*fetchRequest, bool, error) {
    // Short circuit if the pool has been depleted, or if the peer's already
    // downloading something (sanity check not to corrupt state)
    if taskQueue.Empty() {
        return nil, false, nil
    }
    // 如果這個peer還有下載任務沒有完成。
    if _, ok := pendPool[p.id]; ok {
        return nil, false, nil
    }
    // Calculate an upper limit on the items we might fetch (i.e. throttling)
    // 計算我們需要獲取的上限。
    space := len(q.resultCache) - len(donePool)
    // 還需要減去正在下載的數量。
    for _, request := range pendPool {
        space -= len(request.Headers)
    }
    // Retrieve a batch of tasks, skipping previously failed ones
    send := make([]*types.Header, 0, count)
    skip := make([]*types.Header, 0)

    progress := false
    for proc := 0; proc < space && len(send) < count && !taskQueue.Empty(); proc++ {
        header := taskQueue.PopItem().(*types.Header)

        // If we're the first to request this task, initialise the result container
        index := int(header.Number.Int64() - int64(q.resultOffset))
        // index 是結果應該儲存在resultCache的哪一部分。
        if index >= len(q.resultCache) || index < 0 {
            common.Report("index allocation went beyond available resultCache space")
            return nil, false, errInvalidChain
        }
        if q.resultCache[index] == nil { // 第一次排程 有可能多次排程。 那這裡可能就是非空的。
            components := 1
            if q.mode == FastSync && header.Number.Uint64() <= q.fastSyncPivot {
                // 如果是快速同步，那麼需要下載的元件還有 收據receipt
                components = 2
            }
            q.resultCache[index] = &fetchResult{
                Pending: components,
                Header:  header,
            }
        }
        // If this fetch task is a noop, skip this fetch operation
        if isNoop(header) {
            // 如果header的區塊中沒有包含交易，那麼不需要獲取區塊頭
            donePool[header.Hash()] = struct{}{}
            delete(taskPool, header.Hash())

            space, proc = space-1, proc-1
            q.resultCache[index].Pending--
            progress = true
            continue
        }
        // Otherwise unless the peer is known not to have the data, add to the retrieve list
        // Lacks代表節點之前明確表示過沒有這個hash的資料。
        if p.Lacks(header.Hash()) {
            skip = append(skip, header)
        } else {
            send = append(send, header)
        }
    }
    // Merge all the skipped headers back
    for _, header := range skip {
        taskQueue.Push(header, -float32(header.Number.Uint64()))
    }
    if progress {
        // Wake WaitResults, resultCache was modified
        // 通知WaitResults， resultCache有改變
        q.active.Signal()
    }
    // Assemble and return the block download request
    if len(send) == 0 {
        return nil, progress, nil
    }
    request := &fetchRequest{
        Peer:    p,
        Headers: send,
        Time:    time.Now(),
    }
    pendPool[p.id] = request

    return request, progress, nil
}

ReserveReceipts 可以看到和ReserveBodys差不多。不過是佇列換了而已。

// ReserveReceipts reserves a set of receipt fetches for the given peer, skipping
// any previously failed downloads. Beside the next batch of needed fetches, it
// also returns a flag whether empty receipts were queued requiring importing.
func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bool, error) {
    isNoop := func(header *types.Header) bool {
        return header.ReceiptHash == types.EmptyRootHash
    }
    q.lock.Lock()
    defer q.lock.Unlock()

    return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, isNoop)
}

DeliverXXX

Deliver方法在資料下載完之後會被呼叫。

// DeliverBodies injects a block body retrieval response into the results queue.
// The method returns the number of blocks bodies accepted from the delivery and
// also wakes any threads waiting for data delivery.
// DeliverBodies把一個 請求區塊體的返回值插入到results佇列
// 這個方法返回被delivery的區塊體數量，同時會喚醒等待資料的執行緒
func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLists [][]*types.Header) (int, error) {
    q.lock.Lock()
    defer q.lock.Unlock()

    reconstruct := func(header *types.Header, index int, result *fetchResult) error {
        if types.DeriveSha(types.Transactions(txLists[index])) != header.TxHash || types.CalcUncleHash(uncleLists[index]) != header.UncleHash {
            return errInvalidBody
        }
        result.Transactions = txLists[index]
        result.Uncles = uncleLists[index]
        return nil
    }
    return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, bodyReqTimer, len(txLists), reconstruct)
}

deliver方法

func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
    pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, reqTimer metrics.Timer,
    results int, reconstruct func(header *types.Header, index int, result *fetchResult) error) (int, error) {

    // Short circuit if the data was never requested
    // 檢查 資料是否從來沒有請求過。
    request := pendPool[id]
    if request == nil {
        return 0, errNoFetchesPending
    }
    reqTimer.UpdateSince(request.Time)
    delete(pendPool, id)

    // If no data items were retrieved, mark them as unavailable for the origin peer
    if results == 0 {
        //如果結果為空。 那麼標識這個peer沒有這些資料。
        for _, header := range request.Headers {
            request.Peer.MarkLacking(header.Hash())
        }
    }
    // Assemble each of the results with their headers and retrieved data parts
    var (
        accepted int
        failure  error
        useful   bool
    )
    for i, header := range request.Headers {
        // Short circuit assembly if no more fetch results are found
        if i >= results {
            break
        }
        // Reconstruct the next result if contents match up
        index := int(header.Number.Int64() - int64(q.resultOffset))
        if index >= len(q.resultCache) || index < 0 || q.resultCache[index] == nil {
            failure = errInvalidChain
            break
        }
        // 呼叫傳入的函式對資料進行構建
        if err := reconstruct(header, i, q.resultCache[index]); err != nil {
            failure = err
            break
        }
        donePool[header.Hash()] = struct{}{}
        q.resultCache[index].Pending--
        useful = true
        accepted++

        // Clean up a successful fetch
        // 從taskPool刪除。加入donePool
        request.Headers[i] = nil
        delete(taskPool, header.Hash())
    }
    // Return all failed or missing fetches to the queue
    // 所有沒有成功的請求加入taskQueue
    for _, header := range request.Headers {
        if header != nil {
            taskQueue.Push(header, -float32(header.Number.Uint64()))
        }
    }
    // Wake up WaitResults
    // 如果結果有變更，通知WaitResults執行緒啟動。
    if accepted > 0 {
        q.active.Signal()
    }
    // If none of the data was good, it's a stale delivery
    switch {
    case failure == nil || failure == errInvalidChain:
        return accepted, failure
    case useful:
        return accepted, fmt.Errorf("partial failure: %v", failure)
    default:
        return accepted, errStaleDelivery
    }
}

ExpireXXX and CancelXXX

ExpireXXX

ExpireBodies函式獲取了鎖，然後直接呼叫了expire函式。

// ExpireBodies checks for in flight block body requests that exceeded a timeout
// allowance, canceling them and returning the responsible peers for penalisation.
func (q *queue) ExpireBodies(timeout time.Duration) map[string]int {
    q.lock.Lock()
    defer q.lock.Unlock()

    return q.expire(timeout, q.blockPendPool, q.blockTaskQueue, bodyTimeoutMeter)
}

expire函式，

// expire is the generic check that move expired tasks from a pending pool back
// into a task pool, returning all entities caught with expired tasks.
// expire是通用檢查，將過期任務從待處理池移回任務池，返回所有捕獲已到期任務的實體。

func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest, taskQueue *prque.Prque, timeoutMeter metrics.Meter) map[string]int {
    // Iterate over the expired requests and return each to the queue
    expiries := make(map[string]int)
    for id, request := range pendPool {
        if time.Since(request.Time) > timeout {
            // Update the metrics with the timeout
            timeoutMeter.Mark(1)

            // Return any non satisfied requests to the pool
            if request.From > 0 {
                taskQueue.Push(request.From, -float32(request.From))
            }
            for hash, index := range request.Hashes {
                taskQueue.Push(hash, float32(index))
            }
            for _, header := range request.Headers {
                taskQueue.Push(header, -float32(header.Number.Uint64()))
            }
            // Add the peer to the expiry report along the the number of failed requests
            expirations := len(request.Hashes)
            if expirations < len(request.Headers) {
                expirations = len(request.Headers)
            }
            expiries[id] = expirations
        }
    }
    // Remove the expired requests from the pending pool
    for id := range expiries {
        delete(pendPool, id)
    }
    return expiries
}

CancelXXX

Cancle函式取消已經分配的任務， 把任務重新加入到任務池。

// CancelBodies aborts a body fetch request, returning all pending headers to the
// task queue.
func (q *queue) CancelBodies(request *fetchRequest) {
    q.cancel(request, q.blockTaskQueue, q.blockPendPool)
}

// Cancel aborts a fetch request, returning all pending hashes to the task queue.
func (q *queue) cancel(request *fetchRequest, taskQueue *prque.Prque, pendPool map[string]*fetchRequest) {
    q.lock.Lock()
    defer q.lock.Unlock()

    if request.From > 0 {
        taskQueue.Push(request.From, -float32(request.From))
    }
    for hash, index := range request.Hashes {
        taskQueue.Push(hash, float32(index))
    }
    for _, header := range request.Headers {
        taskQueue.Push(header, -float32(header.Number.Uint64()))
    }
    delete(pendPool, request.Peer.id)
}

ScheduleSkeleton

Schedule方法傳入的是已經fetch好的header。Schedule(headers []*types.Header, from uint64)。而ScheduleSkeleton函式的引數是一個骨架， 然後請求對骨架進行填充。所謂的骨架是指我首先每隔192個區塊請求一個區塊頭，然後把返回的header傳入ScheduleSkeleton。 在Schedule函式中只需要queue排程區塊體和回執的下載，而在ScheduleSkeleton函式中，還需要排程那些缺失的區塊頭的下載。

// ScheduleSkeleton adds a batch of header retrieval tasks to the queue to fill
// up an already retrieved header skeleton.
func (q *queue) ScheduleSkeleton(from uint64, skeleton []*types.Header) {
    q.lock.Lock()
    defer q.lock.Unlock()

    // No skeleton retrieval can be in progress, fail hard if so (huge implementation bug)
    if q.headerResults != nil {
        panic("skeleton assembly already in progress")
    }
    // Shedule all the header retrieval tasks for the skeleton assembly
    // 因為這個方法在skeleton為false的時候不會呼叫。 所以一些初始化工作放在這裡執行。
    q.headerTaskPool = make(map[uint64]*types.Header)
    q.headerTaskQueue = prque.New()
    q.headerPeerMiss = make(map[string]map[uint64]struct{}) // Reset availability to correct invalid chains
    q.headerResults = make([]*types.Header, len(skeleton)*MaxHeaderFetch)
    q.headerProced = 0
    q.headerOffset = from
    q.headerContCh = make(chan bool, 1)

    for i, header := range skeleton {
        index := from + uint64(i*MaxHeaderFetch)
        // 每隔MaxHeaderFetch這麼遠有一個header
        q.headerTaskPool[index] = header
        q.headerTaskQueue.Push(index, -float32(index))
    }
}

ReserveHeaders

這個方法只skeleton的模式下才會被呼叫。 用來給peer保留fetch 區塊頭的任務。

// ReserveHeaders reserves a set of headers for the given peer, skipping any
// previously failed batches.
func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest {
    q.lock.Lock()
    defer q.lock.Unlock()

    // Short circuit if the peer's already downloading something (sanity check to
    // not corrupt state)
    if _, ok := q.headerPendPool[p.id]; ok {
        return nil
    }
    // Retrieve a batch of hashes, skipping previously failed ones
    // 從佇列中獲取一個，跳過之前失敗過的節點。
    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)
    }
    // Merge all the skipped batches back
    for _, from := range skip {
        q.headerTaskQueue.Push(from, -float32(from))
    }
    // Assemble and return the block download request
    if send == 0 {
        return nil
    }
    request := &fetchRequest{
        Peer: p,
        From: send,
        Time: time.Now(),
    }
    q.headerPendPool[p.id] = request
    return request
}

DeliverHeaders

// DeliverHeaders injects a header retrieval response into the header results
// cache. This method either accepts all headers it received, or none of them
// if they do not map correctly to the skeleton.
// 這個方法對於所有的區塊頭，要麼全部接收，要麼全部拒絕(如果不能對映到一個skeleton上面)
// If the headers are accepted, the method makes an attempt to deliver the set
// of ready headers to the processor to keep the pipeline full. However it will
// not block to prevent stalling other pending deliveries.
// 如果區塊頭被接收，這個方法會試圖把他們投遞到headerProcCh管道上面。 不過這個方法不會阻塞式的投遞。而是嘗試投遞，如果不能投遞就返回。
func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh chan []*types.Header) (int, error) {
    q.lock.Lock()
    defer q.lock.Unlock()

    // Short circuit if the data was never requested
    request := q.headerPendPool[id]
    if request == nil {
        return 0, errNoFetchesPending
    }
    headerReqTimer.UpdateSince(request.Time)
    delete(q.headerPendPool, id)

    // Ensure headers can be mapped onto the skeleton chain
    target := q.headerTaskPool[request.From].Hash()

    accepted := len(headers) == MaxHeaderFetch
    if accepted { //首先長度需要匹配， 然後檢查區塊號和最後一塊區塊的Hash值是否能夠對應上。
        if headers[0].Number.Uint64() != request.From {
            log.Trace("First header broke chain ordering", "peer", id, "number", headers[0].Number, "hash", headers[0].Hash(), request.From)
            accepted = false
        } else if headers[len(headers)-1].Hash() != target {
            log.Trace("Last header broke skeleton structure ", "peer", id, "number", headers[len(headers)-1].Number, "hash", headers[len(headers)-1].Hash(), "expected", 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 {
                log.Warn("Header broke chain ordering", "peer", id, "number", header.Number, "hash", hash, "expected", want)
                accepted = false
                break
            }
            if headers[i].Hash() != header.ParentHash {
                log.Warn("Header broke chain ancestry", "peer", id, "number", header.Number, "hash", hash)
                accepted = false
                break
            }
        }
    }
    // If the batch of headers wasn't accepted, mark as unavailable
    if !accepted { // 如果不被接收，那麼標記這個peer在這個任務上的失敗。下次請求就不會投遞給這個peer
        log.Trace("Skeleton filling not accepted", "peer", id, "from", request.From)

        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, -float32(request.From))
        return 0, errors.New("delivery not accepted")
    }
    // Clean up a successful fetch and try to deliver any sub-results
    copy(q.headerResults[request.From-q.headerOffset:], headers)
    delete(q.headerTaskPool, request.From)

    ready := 0
    for q.headerProced+ready < len(q.headerResults) && q.headerResults[q.headerProced+ready] != nil {//計算這次到來的header可以讓headerResults有多少資料可以投遞了。
        ready += MaxHeaderFetch
    }
    if ready > 0 {
        // Headers are ready for delivery, gather them and push forward (non blocking)
        process := make([]*types.Header, ready)
        copy(process, q.headerResults[q.headerProced:q.headerProced+ready])
        // 嘗試投遞
        select {
        case headerProcCh <- process:
            log.Trace("Pre-scheduled new headers", "peer", id, "count", len(process), "from", process[0].Number)
            q.headerProced += len(process)
        default:
        }
    }
    // Check for termination and return
    if len(q.headerTaskPool) == 0 {
        // 這個通道比較重要， 如果這個通道接收到資料，說明所有的header任務已經完成。
        q.headerContCh <- false
    }
    return len(headers), nil
}

RetrieveHeaders，ScheduleSkeleton函式在上次排程還沒有做完的情況下是不會呼叫的。 所以上次呼叫完成之後，會使用這個方法來獲取結果，重置狀態。

// RetrieveHeaders retrieves the header chain assemble based on the scheduled
// skeleton.
func (q *queue) RetrieveHeaders() ([]*types.Header, int) {
    q.lock.Lock()
    defer q.lock.Unlock()

    headers, proced := q.headerResults, q.headerProced
    q.headerResults, q.headerProced = nil, 0

    return headers, proced
}