diff --git a/turbo/stages/headerdownload/header_algos.go b/turbo/stages/headerdownload/header_algos.go index 185d07bd0fe..6fc5f440f65 100644 --- a/turbo/stages/headerdownload/header_algos.go +++ b/turbo/stages/headerdownload/header_algos.go @@ -230,10 +230,10 @@ func (hd *HeaderDownload) extendDown(segment ChainSegment) (bool, error) { newAnchor, preExisting := hd.anchors[newAnchorHeader.ParentHash] if !preExisting { newAnchor = &Anchor{ - parentHash: newAnchorHeader.ParentHash, - timestamp: 0, - peerID: anchor.peerID, - blockHeight: newAnchorH.Number, + parentHash: newAnchorHeader.ParentHash, + nextRetryTime: 0, // Will ensure this anchor will be top priority + peerID: anchor.peerID, + blockHeight: newAnchorH.Number, } if newAnchor.blockHeight > 0 { hd.anchors[newAnchorHeader.ParentHash] = newAnchor @@ -353,10 +353,10 @@ func (hd *HeaderDownload) newAnchor(segment ChainSegment, peerID enode.ID) (bool return false, fmt.Errorf("too many anchors: %d, limit %d", len(hd.anchors), hd.anchorLimit) } anchor = &Anchor{ - parentHash: anchorHeader.ParentHash, - peerID: peerID, - timestamp: 0, - blockHeight: anchorH.Number, + parentHash: anchorHeader.ParentHash, + peerID: peerID, + nextRetryTime: 0, // Will ensure this anchor will be top priority + blockHeight: anchorH.Number, } hd.anchors[anchorHeader.ParentHash] = anchor heap.Push(hd.anchorQueue, anchor) @@ -590,7 +590,7 @@ func (hd *HeaderDownload) RequestMoreHeaders(currentTime uint64) (*HeaderRequest for hd.anchorQueue.Len() > 0 { anchor := (*hd.anchorQueue)[0] if _, ok := hd.anchors[anchor.parentHash]; ok { - if anchor.timestamp > currentTime { + if anchor.nextRetryTime > currentTime { // Anchor not ready for re-request yet return nil, penalties } @@ -616,7 +616,7 @@ func (hd *HeaderDownload) SentRequest(req *HeaderRequest, currentTime, timeout u return } anchor.timeouts++ - anchor.timestamp = currentTime + timeout + anchor.nextRetryTime = currentTime + timeout heap.Fix(hd.anchorQueue, anchor.idx) } @@ -625,22 +625,22 @@ func (hd *HeaderDownload) RequestSkeleton() *HeaderRequest { defer hd.lock.RUnlock() log.Trace("Request skeleton", "anchors", len(hd.anchors), "top seen height", hd.topSeenHeight, "highestInDb", hd.highestInDb) stride := uint64(8 * 192) - nextHeight := hd.highestInDb + stride - maxHeight := hd.topSeenHeight + 1 // Inclusive upper bound - if maxHeight <= nextHeight { + strideHeight := hd.highestInDb + stride + lowestAnchorHeight := hd.topSeenHeight + 1 // Inclusive upper bound + if lowestAnchorHeight <= strideHeight { return nil } // Determine the query range as the height of lowest anchor for _, anchor := range hd.anchors { - if anchor.blockHeight > nextHeight && anchor.blockHeight < maxHeight { - maxHeight = anchor.blockHeight // Exclusive upper bound + if anchor.blockHeight > strideHeight && anchor.blockHeight < lowestAnchorHeight { + lowestAnchorHeight = anchor.blockHeight // Exclusive upper bound } } - length := (maxHeight - nextHeight) / stride + length := (lowestAnchorHeight - strideHeight) / stride if length > 192 { length = 192 } - return &HeaderRequest{Number: nextHeight, Length: length, Skip: stride - 1, Reverse: false} + return &HeaderRequest{Number: strideHeight, Length: length, Skip: stride - 1, Reverse: false} } // InsertHeaders attempts to insert headers into the database, verifying them first diff --git a/turbo/stages/headerdownload/header_data_struct.go b/turbo/stages/headerdownload/header_data_struct.go index b6bc67e2b75..a6c12094abf 100644 --- a/turbo/stages/headerdownload/header_data_struct.go +++ b/turbo/stages/headerdownload/header_data_struct.go @@ -75,16 +75,35 @@ func (lq *LinkQueue) Pop() interface{} { return x } +// Anchor represents a header that we do not yet have, but that we would like to have soon +// anchors are created when we know the hash of the header (from its children), and we can +// also derive its blockheight (also from its children), but the corresponding header +// either has not been requested yet, or has not been delivered yet. +// It is possible for one anchor to reference multiple links, because more than one +// header may share the same parent header. In such cases, `links` field will contain +// more than one pointer. type Anchor struct { - peerID enode.ID - links []*Link // Links attached immediately to this anchor - parentHash common.Hash // Hash of the header this anchor can be connected to (to disappear) - blockHeight uint64 - timestamp uint64 // Zero when anchor has just been created, otherwise timestamps when timeout on this anchor request expires - timeouts int // Number of timeout that this anchor has experiences - after certain threshold, it gets invalidated - idx int // Index of the anchor in the queue to be able to modify specific items + peerID enode.ID + links []*Link // Links attached immediately to this anchor + parentHash common.Hash // Hash of the header this anchor can be connected to (to disappear) + blockHeight uint64 + nextRetryTime uint64 // Zero when anchor has just been created, otherwise time when anchor needs to be check to see if retry is neeeded + timeouts int // Number of timeout that this anchor has experiences - after certain threshold, it gets invalidated + idx int // Index of the anchor in the queue to be able to modify specific items } +// AnchorQueue is a priority queue of anchors that priorises by the time when +// another retry on the extending the anchor needs to be attempted +// Every time anchor's extension is requested, the `nextRetryTime` is reset +// to 5 seconds in the future, and when it expires, and the anchor is still +// retry is made +// It implement heap.Interface to be useable by the standard library `heap` +// as a priority queue (implemented as a binary heap) +// As anchors are moved around in the binary heap, they internally track their +// position in the heap (using `idx` field). This feature allows updating +// the heap (using `Fix` function) in situations when anchor is accessed not +// throught the priority queue, but through the map `anchor` in the +// HeaderDownloader type. type AnchorQueue []*Anchor func (aq AnchorQueue) Len() int { @@ -92,11 +111,11 @@ func (aq AnchorQueue) Len() int { } func (aq AnchorQueue) Less(i, j int) bool { - if aq[i].timestamp == aq[j].timestamp { - // When timestamps are the same, we prioritise low block height anchors + if aq[i].nextRetryTime == aq[j].nextRetryTime { + // When next retry times are the same, we prioritise low block height anchors return aq[i].blockHeight < aq[j].blockHeight } - return aq[i].timestamp < aq[j].timestamp + return aq[i].nextRetryTime < aq[j].nextRetryTime } func (aq AnchorQueue) Swap(i, j int) {