diff --git a/turbo/stages/headerdownload/header_algos.go b/turbo/stages/headerdownload/header_algos.go
index 185d07bd0fec8aa1a196fdd89967db6ff11f20eb..6fc5f440f6546580305ff989c0214427d2be3fd0 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 b6bc67e2b751eb1c512ae64c5aa6aedf61ed612c..a6c12094abf3f298cd131bc98edfb693380d4212 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) {