queueWithHeaders.DeliverHeaders¶
| Source (upstream pre-purge) | Current | |
|---|---|---|
| File | eth/downloader/queue.go |
queue_pow.go |
| Symbol | queue.DeliverHeaders |
queueWithHeaders.DeliverHeaders |
| Ref | 45baf2111~1 |
etc/v1.17.3-full-node |
3-way merge — purge → getc ← upstream¶
pre-purge≈ adapted (origin inferred by similarity)fork-only
↗// 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.
//
// 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.
func (q *queueWithHeaders) DeliverHeaders(id string, headers []*types.Header, hashes []common.Hash, headerProcCh chan *headerTask) (int, error) {
≈ q.queue.lock.Lock()
defer q.queue.lock.Unlock()
↗ var logger log.Logger
if len(id) < 16 {
// Tests use short IDs, don't choke on them
logger = log.New("peer", id)
} else {
logger = log.New("peer", id[:16])
}
// Short circuit if the data was never requested
request := q.headerPendPool[id]
if request == nil {
headerDropMeter.Mark(int64(len(headers)))
return 0, errNoFetchesPending
}
delete(q.headerPendPool, id)
↗ headerReqTimer.UpdateSince(request.Time)
headerInMeter.Mark(int64(len(headers)))
↗ // Ensure headers can be mapped onto the skeleton chain
target := q.headerTaskPool[request.From].Hash()
↗ accepted := len(headers) == MaxHeaderFetch
if accepted {
if headers[0].Number.Uint64() != request.From {
logger.Trace("First header broke chain ordering", "number", headers[0].Number, "hash", hashes[0], "expected", request.From)
accepted = false
} else if hashes[len(headers)-1] != target {
logger.Trace("Last header broke skeleton structure ", "number", headers[len(headers)-1].Number, "hash", hashes[len(headers)-1], "expected", target)
accepted = false
}
}
if accepted {
parentHash := hashes[0]
for i, header := range headers[1:] {
hash := hashes[i+1]
if want := request.From + 1 + uint64(i); header.Number.Uint64() != want {
logger.Warn("Header broke chain ordering", "number", header.Number, "hash", hash, "expected", want)
accepted = false
break
}
if parentHash != header.ParentHash {
logger.Warn("Header broke chain ancestry", "number", header.Number, "hash", hash)
accepted = false
break
}
// Set-up parent hash for next round
parentHash = hash
}
}
// If the batch of headers wasn't accepted, mark as unavailable
if !accepted {
logger.Trace("Skeleton filling not accepted", "from", request.From)
headerDropMeter.Mark(int64(len(headers)))
↗ 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")
}
// Clean up a successful fetch and try to deliver any sub-results
copy(q.headerResults[request.From-q.headerOffset:], headers)
copy(q.headerHashes[request.From-q.headerOffset:], hashes)
↗ delete(q.headerTaskPool, request.From)
↗ ready := 0
for q.headerProced+ready < len(q.headerResults) && q.headerResults[q.headerProced+ready] != nil {
ready += MaxHeaderFetch
}
if ready > 0 {
// Headers are ready for delivery, gather them and push forward (non blocking)
processHeaders := make([]*types.Header, ready)
copy(processHeaders, q.headerResults[q.headerProced:q.headerProced+ready])
↗ processHashes := make([]common.Hash, ready)
copy(processHashes, q.headerHashes[q.headerProced:q.headerProced+ready])
↗ select {
case headerProcCh <- &headerTask{
headers: processHeaders,
hashes: processHashes,
}:
logger.Trace("Pre-scheduled new headers", "count", len(processHeaders), "from", processHeaders[0].Number)
q.headerProced += len(processHeaders)
default:
}
}
// Check for termination and return
if len(q.headerTaskPool) == 0 {
q.headerContCh <- false
}
return len(headers), nil
}
core-geth validation — +3 -3
| | | |---|---| | File | [`queue.go`](https://github.com/etclabscore/core-geth/blob/v1.12.20/eth/downloader/queue.go) | | Symbol | `queue.DeliverHeaders` | | Ref | `v1.12.20` |--- a/core-geth/eth/downloader/queue.go
+++ b/etc/eth/downloader/queue_pow.go
@@ -5,9 +5,9 @@
// 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.
-func (q *queue) DeliverHeaders(id string, headers []*types.Header, hashes []common.Hash, headerProcCh chan *headerTask) (int, error) {
- q.lock.Lock()
- defer q.lock.Unlock()
+func (q *queueWithHeaders) DeliverHeaders(id string, headers []*types.Header, hashes []common.Hash, headerProcCh chan *headerTask) (int, error) {
+ q.queue.lock.Lock()
+ defer q.queue.lock.Unlock()
var logger log.Logger
if len(id) < 16 {