diff --git a/core/types/block.go b/core/types/block.go index 25823b1653..8129a7f6f9 100644 --- a/core/types/block.go +++ b/core/types/block.go @@ -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 } diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 25830239ef..604727efb7 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -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 { diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index 58e544796a..fbb6d2312d 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -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() } diff --git a/eth/downloader/metrics.go b/eth/downloader/metrics.go index 518ffcc877..c38732043a 100644 --- a/eth/downloader/metrics.go +++ b/eth/downloader/metrics.go @@ -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) ) diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index f0dfff5283..4d2edaa43b 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -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) { diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 390900711b..425f249ba1 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -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++