diff --git a/core/types/block.go b/core/types/block.go index 8129a7f6f9..de4db0fed6 100644 --- a/core/types/block.go +++ b/core/types/block.go @@ -137,12 +137,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 +// that is: no transactions and no uncles. func (h *Header) EmptyBody() bool { return h.TxHash == EmptyRootHash && h.UncleHash == EmptyUncleHash } -// EmptyReceipts returns true if there are no receipts for this header/block +// EmptyReceipts returns true if there are no receipts for this header/block. func (h *Header) EmptyReceipts() bool { return h.ReceiptHash == EmptyRootHash } diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 9bbafd083b..5e9ce44a85 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -511,9 +511,9 @@ func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td *big.I d.syncInitHook(origin, height) } fetchers := []func() error{ - func() error { return d.fetchHeaders(p, origin+1, pivot, height) }, // Headers are always retrieved - func() error { return d.fetchBodies(origin + 1) }, // Bodies are retrieved during normal and fast sync - func() error { return d.fetchReceipts(origin + 1) }, // Receipts are retrieved during fast sync + func() error { return d.fetchHeaders(p, origin+1, pivot) }, // Headers are always retrieved + func() error { return d.fetchBodies(origin + 1) }, // Bodies are retrieved during normal and fast sync + func() error { return d.fetchReceipts(origin + 1) }, // Receipts are retrieved during fast sync func() error { return d.processHeaders(origin+1, pivot, td) }, } if d.mode == FastSync { @@ -621,7 +621,7 @@ func (d *Downloader) fetchHeight(p *peerConnection) (*types.Header, error) { // Make sure the peer actually gave something valid headers := packet.(*headerPack).headers if len(headers) != 1 { - p.log.Info("Multiple headers for single request", "headers", len(headers)) + p.log.Warn("Multiple headers for single request", "headers", len(headers)) return nil, errBadPeer } head := headers[0] @@ -853,7 +853,7 @@ func (d *Downloader) findAncestor(p *peerConnection, remoteHeader *types.Header) // Make sure the peer actually gave something valid headers := packer.(*headerPack).headers if len(headers) != 1 { - p.log.Info("Multiple headers for single request", "headers", len(headers)) + p.log.Warn("Multiple headers for single request", "headers", len(headers)) return 0, errBadPeer } arrived = true @@ -877,7 +877,7 @@ func (d *Downloader) findAncestor(p *peerConnection, remoteHeader *types.Header) } header := d.lightchain.GetHeaderByHash(h) // Independent of sync mode, header surely exists if header.Number.Uint64() != check { - p.log.Info("Received non requested header", "number", header.Number, "hash", header.Hash(), "request", check) + p.log.Warn("Received non requested header", "number", header.Number, "hash", header.Hash(), "request", check) return 0, errBadPeer } start = check @@ -910,7 +910,7 @@ func (d *Downloader) findAncestor(p *peerConnection, remoteHeader *types.Header) // other peers are only accepted if they map cleanly to the skeleton. If no one // can fill in the skeleton - not even the origin peer - it's assumed invalid and // the origin is dropped. -func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead uint64) error { +func (d *Downloader) fetchHeaders(p *peerConnection, from uint64, pivot uint64) error { p.log.Debug("Directing header downloads", "origin", from) defer p.log.Debug("Header download terminated") @@ -927,6 +927,7 @@ func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead ttl = d.requestTTL() timeout.Reset(ttl) + if skeleton { p.log.Trace("Fetching skeleton headers", "count", MaxHeaderFetch, "from", from) go p.peer.RequestHeadersByNumber(from+uint64(MaxHeaderFetch)-1, MaxSkeletonSize, MaxHeaderFetch-1, false) @@ -982,7 +983,7 @@ func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead } } headers := packet.(*headerPack).headers - ignoredHeaders := 0 + // If we received a skeleton batch, resolve internals concurrently if skeleton { filled, proced, err := d.fillHeaderSkeleton(from, headers) @@ -1020,7 +1021,6 @@ func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead delay = n } headers = headers[:n-delay] - ignoredHeaders = delay } } } @@ -1035,15 +1035,8 @@ func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead from += uint64(len(headers)) getHeaders(from) } else { - // No headers delivered - if !skeleton && ignoredHeaders == 0 && from < advertisedHead { - // This peer told is about a high number, but is not - // delivering - p.log.Trace("peer did not deliver", "requested", from, "head", advertisedHead) - return errStallingPeer - } - p.log.Trace("All headers delayed, waiting") // No headers delivered, or all of them being delayed, sleep a bit and retry + p.log.Trace("All headers delayed, waiting") select { case <-time.After(fsHeaderContCheck): getHeaders(from) @@ -1328,7 +1321,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack) // have them. request, progress, throttle, err := reserve(peer, capacity(peer)) if err != nil { - log.Info("Error in loop", "err", err) + log.Info("Error reserving fetch", "err", err) return err } if progress { @@ -1489,7 +1482,8 @@ func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) er rollbackErr = err rollback = append(rollback, chunk[:n]...) } - log.Info("Invalid header encountered", "number", chunk[n].Number, "hash", chunk[n].Hash(), "err", err, "parent", chunk[n].ParentHash) + log.Info("Invalid header encountered", "number", chunk[n].Number, + "hash", chunk[n].Hash(), "parent", chunk[n].ParentHash, "err", err) return errInvalidChain } // All verifications passed, store newly found uncertain headers @@ -1682,12 +1676,11 @@ func (d *Downloader) processFastSyncContent(latest *types.Header) error { func splitAroundPivot(pivot uint64, results []*fetchResult) (p *fetchResult, before, after []*fetchResult) { if len(results) == 0 { - return + return nil, nil, nil } if lastNum := results[len(results)-1].Header.Number.Uint64(); lastNum < pivot { // the pivot is somewhere in the future - before = results - return + return nil, results, nil } // This can also be optimized, but only happens very seldom for _, result := range results { diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index 0fa83a70a4..a421799f09 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -115,15 +115,11 @@ func (w *lightPeerWrapper) RequestNodeData([]common.Hash) error { // newPeerConnection creates a new downloader peer. func newPeerConnection(id string, version int, peer Peer, logger log.Logger) *peerConnection { return &peerConnection{ - id: id, - lacking: make(map[common.Hash]struct{}), - peer: peer, - version: version, - log: logger, - headerThroughput: float64(MaxHeaderFetch / 8), - blockThroughput: float64(MaxBlockFetch / 8), - receiptThroughput: float64(MaxReceiptFetch / 8), - stateThroughput: float64(MaxStateFetch / 8), + id: id, + lacking: make(map[common.Hash]struct{}), + peer: peer, + version: version, + log: logger, } } @@ -137,10 +133,10 @@ func (p *peerConnection) Reset() { atomic.StoreInt32(&p.receiptIdle, 0) atomic.StoreInt32(&p.stateIdle, 0) - p.headerThroughput = float64(MaxHeaderFetch / 8) - p.blockThroughput = float64(MaxBlockFetch / 8) - p.receiptThroughput = float64(MaxReceiptFetch / 8) - p.stateThroughput = float64(MaxStateFetch / 8) + p.headerThroughput = 0 + p.blockThroughput = 0 + p.receiptThroughput = 0 + p.stateThroughput = 0 p.lacking = make(map[common.Hash]struct{}) } diff --git a/eth/downloader/peer_test.go b/eth/downloader/peer_test.go index 2bb6291887..4bf0e200bb 100644 --- a/eth/downloader/peer_test.go +++ b/eth/downloader/peer_test.go @@ -1,3 +1,19 @@ +// Copyright 2020 The go-ethereum Authors +// This file is part of go-ethereum. +// +// go-ethereum is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// go-ethereum is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with go-ethereum. If not, see . + package downloader import ( diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 0fa8336dfa..0e033482af 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -34,8 +34,8 @@ import ( ) const ( - BodyType = 0 - ReceiptType = 1 + bodyType = uint(0) + receiptType = uint(1) ) var ( @@ -73,44 +73,37 @@ func newFetchResult(header *types.Header, fastSync bool) *fetchResult { Header: header, } if !header.EmptyBody() { - item.pending = 1 + item.pending |= (1 << bodyType) } if fastSync && !header.EmptyReceipts() { - item.pending += 2 + item.pending |= (1 << receiptType) } return item } // SetBodyDone flags the body as finished. func (f *fetchResult) SetBodyDone() { - if v := atomic.LoadInt32(&f.pending); (v & 1) != 0 { + if v := atomic.LoadInt32(&f.pending); (v & (1 << bodyType)) != 0 { atomic.AddInt32(&f.pending, -1) } } -// AllDone checks item is done +// AllDone checks if item is done. func (f *fetchResult) AllDone() bool { return atomic.LoadInt32(&f.pending) == 0 } // SetReceiptsDone flags the receipts as finished. func (f *fetchResult) SetReceiptsDone() { - if v := atomic.LoadInt32(&f.pending); (v & 2) != 0 { + if v := atomic.LoadInt32(&f.pending); (v & (1 << receiptType)) != 0 { atomic.AddInt32(&f.pending, -2) } } // Done checks if the given type is done already -func (f *fetchResult) Done(typ int) bool { +func (f *fetchResult) Done(kind uint) bool { v := atomic.LoadInt32(&f.pending) - switch typ { - case BodyType: - return (v & 1) == 0 - case ReceiptType: - return (v & 2) == 0 - default: - return false - } + return v&(1< 0 {