forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

eth/downloader: fix up some review concerns

Browse source
This commit is contained in:
Martin Holst Swende 2019-11-04 11:44:42 +01:00
parent 43946c6beb
commit 6804284f31
No known key found for this signature in database
GPG key ID: 683B438C05A5DDF0
6 changed files with 19 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
core/types/block.go
View file

@ -138,12 +138,12 @@ func rlpHash(x interface{}) (h common.Hash) {
// EmptyBody returns true if there is no additional 'body' to complete the header
// that is: no transactions and no uncles
func (h *Header) EmptyBody() bool{
func (h *Header) EmptyBody() bool {
return h.TxHash == EmptyRootHash && h.UncleHash == EmptyUncleHash
}
// EmptyReceipts returns true if there are no receipts for this header/block
func (h *Header) EmptyReceipts() bool{
func (h *Header) EmptyReceipts() bool {
return h.ReceiptHash == EmptyRootHash
}

4
eth/downloader/downloader.go
View file

@ -1191,7 +1191,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
idle func() ([]*peerConnection, int), setIdle func(*peerConnection, int, time.Time), kind string) error {
// Create a ticker to detect expired retrieval tasks
ticker := time.NewTicker(200 * time.Millisecond)
ticker := time.NewTicker(100 * time.Millisecond)
defer ticker.Stop()
update := make(chan struct{}, 1)
@ -1328,7 +1328,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
}
if throttle {
throttled = true
throttleBlockCounter.Inc(1)
throttleCounter.Inc(1)
}
if request != nil {
if request.From > 0 {

2
eth/downloader/downloader_test.go
View file

@ -19,7 +19,6 @@ package downloader
import (
"errors"
"fmt"
"github.com/ethereum/go-ethereum/log"
"math/big"
"strings"
"sync"
@ -735,7 +734,6 @@ func testBoundedHeavyForkedSync(t *testing.T, protocol int, mode SyncMode) {
if err := tester.sync("heavy-rewriter", nil, mode); err != errInvalidAncestor {
t.Fatalf("sync failure mismatch: have %v, want %v", err, errInvalidAncestor)
}
fmt.Printf("terminating\n")
tester.terminate()
}

3
eth/downloader/metrics.go
View file

@ -41,6 +41,5 @@ var (
stateInMeter = metrics.NewRegisteredMeter("eth/downloader/states/in", nil)
stateDropMeter = metrics.NewRegisteredMeter("eth/downloader/states/drop", nil)
throttleBlockCounter = metrics.NewRegisteredCounter("eth/downloader/throttle/blocks", nil)
throttleReceiptCounter = metrics.NewRegisteredCounter("eth/downloader/throttle/receipts", nil)
throttleCounter = metrics.NewRegisteredCounter("eth/downloader/throttle", nil)
)

6
eth/downloader/peer.go
View file

@ -477,12 +477,6 @@ func (ps *peerSet) HeaderIdlePeers() ([]*peerConnection, int) {
return ps.idlePeers(62, 65, idle, throughput)
}
func fullyIdle(p *peerConnection) bool {
return atomic.LoadInt32(&p.blockIdle) == 0 &&
atomic.LoadInt32(&p.receiptIdle) == 0 &&
atomic.LoadInt32(&p.stateIdle) == 0
}
// BodyIdlePeers retrieves a flat list of all the currently body-idle peers within
// the active peer set, ordered by their reputation.
func (ps *peerSet) BodyIdlePeers() ([]*peerConnection, int) {

56
eth/downloader/queue.go
View file

@ -139,9 +139,8 @@ type queue struct {
receiptTaskQueue *prque.Prque // [eth/63] Priority queue of the headers to fetch the receipts for
receiptPendPool map[string]*fetchRequest // [eth/63] Currently pending receipt retrieval operations
resultCache *resultStore // Downloaded but not yet delivered fetch results
//resultOffset uint64 // Offset of the first cached fetch result in the block chain
resultSize common.StorageSize // Approximate size of a block (exponential moving average)
resultCache *resultStore // Downloaded but not yet delivered fetch results
resultSize common.StorageSize // Approximate size of a block (exponential moving average)
lock *sync.RWMutex
active *sync.Cond
@ -304,8 +303,6 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
q.lock.Lock()
defer q.lock.Unlock()
// if the resultCache pushes back, we can stop trying to shove things in there for now
//var pushBack error
// Insert all the headers prioritised by the contained block number
inserts := make([]*types.Header, 0, len(headers))
for _, header := range headers {
@ -319,49 +316,24 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
log.Warn("Header broke chain ancestry", "number", header.Number, "hash", hash)
break
}
// Make sure no duplicate requests are executed
if _, ok := q.blockTaskPool[hash]; ok {
log.Warn("Header already scheduled for block fetch", "number", header.Number, "hash", hash)
continue
if !header.EmptyBody() {
// Make sure no duplicate requests are executed
if _, ok := q.blockTaskPool[hash]; ok {
log.Warn("Header already scheduled for block fetch", "number", header.Number, "hash", hash)
} else {
q.blockTaskPool[hash] = header
q.blockTaskQueue.Push(header, -int64(header.Number.Uint64()))
}
}
q.blockTaskPool[hash] = header
q.blockTaskQueue.Push(header, -int64(header.Number.Uint64()))
// Queue for receipt retrieval
if q.mode == FastSync {
if q.mode == FastSync && !header.EmptyReceipts() {
if _, ok := q.receiptTaskPool[hash]; ok {
log.Warn("Header already scheduled for receipt fetch", "number", header.Number, "hash", hash)
continue
} else {
q.receiptTaskPool[hash] = header
q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64()))
}
q.receiptTaskPool[hash] = header
q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64()))
}
//var bodyNeeded = !header.EmptyBody()
//var receiptNeeded = q.mode == FastSync && !header.EmptyReceipts()
//if pushBack == nil {
// bodyNeeded, receiptNeeded, _, pushBack = q.resultCache.AddFetch(header, q.mode == FastSync)
//}
//if !receiptNeeded {
// bodyNeeded = true
//}
// Queue for body retrieval - unless empty block
//if bodyNeeded {
// q.blockTaskPool[hash] = header
// q.blockTaskQueue.Push(header, -int64(header.Number.Uint64()))
//} else { // otherwise, straight to done
// q.blockDonePool[hash] = struct{}{}
//}
//if receiptNeeded {
// // Queue for receipt retrieval
// if _, ok := q.receiptTaskPool[hash]; ok {
// log.Warn("Header already scheduled for receipt fetch", "number", header.Number, "hash", hash)
// continue
// }
// q.receiptTaskPool[hash] = header
// q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64()))
//} else { // done already
// q.receiptDonePool[hash] = struct{}{}
//}
inserts = append(inserts, header)
q.headerHead = hash
from++