From 1fe952ed44114153808726030960294cf9a1364f Mon Sep 17 00:00:00 2001 From: Martin Holst Swende Date: Sat, 26 Oct 2019 21:45:52 +0200 Subject: [PATCH] downloader, fetcher: throttle-metrics, fetcher filter improvements, standalone resultcache --- core/types/block.go | 11 ++ eth/downloader/downloader.go | 25 +-- eth/downloader/metrics.go | 3 + eth/downloader/peer.go | 73 +++++--- eth/downloader/peer_test.go | 37 ++++ eth/downloader/queue.go | 307 ++++++++++++++++++---------------- eth/downloader/resultcache.go | 258 ++++++++++++++++++++++++++++ eth/fetcher/block_fetcher.go | 69 ++++---- 8 files changed, 577 insertions(+), 206 deletions(-) create mode 100644 eth/downloader/peer_test.go create mode 100644 eth/downloader/resultcache.go diff --git a/core/types/block.go b/core/types/block.go index 741ff8e282..25823b1653 100644 --- a/core/types/block.go +++ b/core/types/block.go @@ -136,6 +136,17 @@ func rlpHash(x interface{}) (h common.Hash) { return h } +// 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{ + return h.TxHash == EmptyRootHash && h.UncleHash == EmptyUncleHash +} + +// EmptyReceipts returns true if there are no receipts for this header/block +func (h *Header) EmptyReceipts() bool{ + return h.ReceiptHash == EmptyRootHash +} + // Body is a simple (mutable, non-safe) data container for storing and moving // a block's data contents (transactions and uncles) together. type Body struct { diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index f8982f696f..bd88bf63b4 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -219,7 +219,7 @@ func New(checkpoint uint64, stateDb ethdb.Database, stateBloom *trie.SyncBloom, stateBloom: stateBloom, mux: mux, checkpoint: checkpoint, - queue: newQueue(), + queue: newQueue(blockCacheItems), peers: newPeerSet(), rttEstimate: uint64(rttMaxEstimate), rttConfidence: uint64(1000000), @@ -619,7 +619,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.Debug("Multiple headers for single request", "headers", len(headers)) + p.log.Info("Multiple headers for single request", "headers", len(headers)) return nil, errBadPeer } head := headers[0] @@ -851,7 +851,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.Debug("Multiple headers for single request", "headers", len(headers)) + p.log.Info("Multiple headers for single request", "headers", len(headers)) return 0, errBadPeer } arrived = true @@ -875,7 +875,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.Debug("Received non requested header", "number", header.Number, "hash", header.Hash(), "request", check) + p.log.Info("Received non requested header", "number", header.Number, "hash", header.Hash(), "request", check) return 0, errBadPeer } start = check @@ -1120,7 +1120,7 @@ func (d *Downloader) fetchBodies(from uint64) error { pack := packet.(*bodyPack) return d.queue.DeliverBodies(pack.peerID, pack.transactions, pack.uncles) } - expire = func() map[string]int { return d.queue.ExpireBodies(d.requestTTL()) } + expire = func() map[string]int {return d.queue.ExpireBodies(d.requestTTL())} fetch = func(p *peerConnection, req *fetchRequest) error { return p.FetchBodies(req) } capacity = func(p *peerConnection) int { return p.BlockCapacity(d.requestRTT()) } setIdle = func(p *peerConnection, accepted int) { p.SetBodiesIdle(accepted) } @@ -1188,7 +1188,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack) idle func() ([]*peerConnection, int), setIdle func(*peerConnection, int), kind string) error { // Create a ticker to detect expired retrieval tasks - ticker := time.NewTicker(100 * time.Millisecond) + ticker := time.NewTicker(200 * time.Millisecond) defer ticker.Stop() update := make(chan struct{}, 1) @@ -1264,7 +1264,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack) // The reason the minimum threshold is 2 is because the downloader tries to estimate the bandwidth // and latency of a peer separately, which requires pushing the measures capacity a bit and seeing // how response times reacts, to it always requests one more than the minimum (i.e. min 2). - if fails > 2 { + if fails > 8 { peer.log.Trace("Data delivery timed out", "type", kind) setIdle(peer, 0) } else { @@ -1323,6 +1323,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack) progressed = true } if request == nil { + //peer.log.Info("no request allocated this loop", "type", kind) continue } if request.From > 0 { @@ -1359,6 +1360,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack) func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) error { // Keep a count of uncertain headers to roll back var rollback []*types.Header + var rollbackErr error defer func() { if len(rollback) > 0 { // Flatten the headers and roll them back @@ -1380,7 +1382,7 @@ func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) er log.Warn("Rolled back headers", "count", len(hashes), "header", fmt.Sprintf("%d->%d", lastHeader, d.lightchain.CurrentHeader().Number), "fast", fmt.Sprintf("%d->%d", lastFastBlock, curFastBlock), - "block", fmt.Sprintf("%d->%d", lastBlock, curBlock)) + "block", fmt.Sprintf("%d->%d", lastBlock, curBlock), "reason", rollbackErr) } }() @@ -1469,9 +1471,10 @@ func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) er if n, err := d.lightchain.InsertHeaderChain(chunk, frequency); err != nil { // If some headers were inserted, add them too to the rollback list if n > 0 { + rollbackErr = err rollback = append(rollback, chunk[:n]...) } - log.Debug("Invalid header encountered", "number", chunk[n].Number, "hash", chunk[n].Hash(), "err", err) + log.Info("Invalid header encountered", "number", chunk[n].Number, "hash", chunk[n].Hash(), "err", err) return errInvalidChain } // All verifications passed, store newly found uncertain headers @@ -1493,7 +1496,7 @@ func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) er // Otherwise insert the headers for content retrieval inserts := d.queue.Schedule(chunk, origin) if len(inserts) != len(chunk) { - log.Debug("Stale headers") + rollbackErr = fmt.Errorf("stale headers: len inserts %v len(chunk) %v", len(inserts), len(chunk)) return errBadPeer } } @@ -1693,7 +1696,7 @@ func (d *Downloader) commitFastSyncData(results []*fetchResult, stateSync *state } // Retrieve the a batch of results to import first, last := results[0].Header, results[len(results)-1].Header - log.Debug("Inserting fast-sync blocks", "items", len(results), + log.Info("Inserting fast-sync blocks", "items", len(results), "firstnum", first.Number, "firsthash", first.Hash(), "lastnumn", last.Number, "lasthash", last.Hash(), ) diff --git a/eth/downloader/metrics.go b/eth/downloader/metrics.go index d4eb337946..518ffcc877 100644 --- a/eth/downloader/metrics.go +++ b/eth/downloader/metrics.go @@ -40,4 +40,7 @@ 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) ) diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index 5c2020d7d8..d9fff2cb7e 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -115,13 +115,15 @@ 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, + 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), } } @@ -135,10 +137,10 @@ func (p *peerConnection) Reset() { atomic.StoreInt32(&p.receiptIdle, 0) atomic.StoreInt32(&p.stateIdle, 0) - p.headerThroughput = 0 - p.blockThroughput = 0 - p.receiptThroughput = 0 - p.stateThroughput = 0 + p.headerThroughput = float64(MaxHeaderFetch / 8) + p.blockThroughput = float64(MaxBlockFetch / 8) + p.receiptThroughput = float64(MaxReceiptFetch / 8) + p.stateThroughput = float64(MaxStateFetch / 8) p.lacking = make(map[common.Hash]struct{}) } @@ -283,7 +285,7 @@ func (p *peerConnection) HeaderCapacity(targetRTT time.Duration) int { p.lock.RLock() defer p.lock.RUnlock() - return int(math.Min(1+math.Max(1, p.headerThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxHeaderFetch))) + return int(math.Min(7+math.Max(1, p.headerThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxHeaderFetch))) } // BlockCapacity retrieves the peers block download allowance based on its @@ -292,7 +294,7 @@ func (p *peerConnection) BlockCapacity(targetRTT time.Duration) int { p.lock.RLock() defer p.lock.RUnlock() - return int(math.Min(1+math.Max(1, p.blockThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxBlockFetch))) + return int(math.Min(7+math.Max(1, p.blockThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxBlockFetch))) } // ReceiptCapacity retrieves the peers receipt download allowance based on its @@ -301,7 +303,7 @@ func (p *peerConnection) ReceiptCapacity(targetRTT time.Duration) int { p.lock.RLock() defer p.lock.RUnlock() - return int(math.Min(1+math.Max(1, p.receiptThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxReceiptFetch))) + return int(math.Min(7+math.Max(1, p.receiptThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxReceiptFetch))) } // NodeDataCapacity retrieves the peers state download allowance based on its @@ -310,7 +312,7 @@ func (p *peerConnection) NodeDataCapacity(targetRTT time.Duration) int { p.lock.RLock() defer p.lock.RUnlock() - return int(math.Min(1+math.Max(1, p.stateThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxStateFetch))) + return int(math.Min(7+math.Max(1, p.stateThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxStateFetch))) } // MarkLacking appends a new entity to the set of items (blocks, receipts, states) @@ -473,6 +475,12 @@ 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) { @@ -523,22 +531,20 @@ func (ps *peerSet) idlePeers(minProtocol, maxProtocol int, idleCheck func(*peerC defer ps.lock.RUnlock() idle, total := make([]*peerConnection, 0, len(ps.peers)), 0 + tps := make([]float64, 0, len(ps.peers)) for _, p := range ps.peers { if p.version >= minProtocol && p.version <= maxProtocol { if idleCheck(p) { idle = append(idle, p) + tps = append(tps, throughput(p)) } total++ } } - for i := 0; i < len(idle); i++ { - for j := i + 1; j < len(idle); j++ { - if throughput(idle[i]) < throughput(idle[j]) { - idle[i], idle[j] = idle[j], idle[i] - } - } - } - return idle, total + // And sort them + sortPeers := &peerThroughputSort{idle, tps} + sort.Sort(sortPeers) + return sortPeers.p, total } // medianRTT returns the median RTT of the peerset, considering only the tuning @@ -571,3 +577,24 @@ func (ps *peerSet) medianRTT() time.Duration { } return median } + +// peerThroughputSort implements the Sort interface, and allows for +// sorting a set of peers by their throughput +// The sorted data is with the _highest_ throughput first +type peerThroughputSort struct { + p []*peerConnection + tp []float64 +} + +func (ps *peerThroughputSort) Len() int { + return len(ps.p) +} + +func (ps *peerThroughputSort) Less(i, j int) bool { + return ps.tp[i] > ps.tp[j] +} + +func (ps *peerThroughputSort) Swap(i, j int) { + ps.p[i], ps.p[j] = ps.p[j], ps.p[i] + ps.tp[i], ps.tp[j] = ps.tp[j], ps.tp[i] +} diff --git a/eth/downloader/peer_test.go b/eth/downloader/peer_test.go new file mode 100644 index 0000000000..082cf86610 --- /dev/null +++ b/eth/downloader/peer_test.go @@ -0,0 +1,37 @@ +package downloader + +import ( + "sort" + "testing" +) + +func TestPeerThroughputSorting(t *testing.T){ + a := &peerConnection{ + id:"a", + headerThroughput:1.25, + } + b := &peerConnection{ + id: "b", + headerThroughput:1.21, + } + c := &peerConnection{ + id: "c", + headerThroughput:1.23, + } + + peers := []*peerConnection{a,b,c} + tps := []float64{a.headerThroughput, + b.headerThroughput, c.headerThroughput} + sortPeers := &peerThroughputSort{peers, tps} + sort.Sort(sortPeers) + if got, exp := sortPeers.p[0].id , "a"; got != exp{ + t.Errorf("sort fail, got %v exp %v", got, exp) + } + if got, exp := sortPeers.p[1].id , "c"; got != exp{ + t.Errorf("sort fail, got %v exp %v", got, exp) + } + if got, exp := sortPeers.p[2].id , "b"; got != exp{ + t.Errorf("sort fail, got %v exp %v", got, exp) + } + +} \ No newline at end of file diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index e90a1e7dde..1058285c43 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -54,7 +54,7 @@ type fetchRequest struct { // fetchResult is a struct collecting partial results from data fetchers until // all outstanding pieces complete and the result as a whole can be processed. type fetchResult struct { - Pending int // Number of data fetches still pending + Pending uint8 // Number of data fetches still pending Hash common.Hash // Hash of the header to prevent recalculating Header *types.Header @@ -89,9 +89,9 @@ type queue struct { receiptPendPool map[string]*fetchRequest // [eth/63] Currently pending receipt retrieval operations receiptDonePool map[common.Hash]struct{} // [eth/63] Set of the completed receipt fetches - resultCache []*fetchResult // 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 + //resultOffset uint64 // Offset of the first cached fetch result in the block chain + resultSize common.StorageSize // Approximate size of a block (exponential moving average) lock *sync.RWMutex active *sync.Cond @@ -99,7 +99,7 @@ type queue struct { } // newQueue creates a new download queue for scheduling block retrieval. -func newQueue() *queue { +func newQueue(blockCacheLimit int) *queue { lock := new(sync.RWMutex) return &queue{ headerPendPool: make(map[string]*fetchRequest), @@ -112,9 +112,10 @@ func newQueue() *queue { receiptTaskQueue: prque.New(nil), receiptPendPool: make(map[string]*fetchRequest), receiptDonePool: make(map[common.Hash]struct{}), - resultCache: make([]*fetchResult, blockCacheItems), - active: sync.NewCond(lock), - lock: lock, + //resultCache: make([]*fetchResult, blockCacheItems), + resultCache: newResultStore(blockCacheLimit), + active: sync.NewCond(lock), + lock: lock, } } @@ -139,8 +140,7 @@ func (q *queue) Reset() { q.receiptPendPool = make(map[string]*fetchRequest) q.receiptDonePool = make(map[common.Hash]struct{}) - q.resultCache = make([]*fetchResult, blockCacheItems) - q.resultOffset = 0 + q.resultCache = newResultStore(blockCacheItems) } // Close marks the end of the sync, unblocking Results. @@ -219,18 +219,24 @@ func (q *queue) Idle() bool { // fetches exceed block cache). func (q *queue) ShouldThrottleBlocks() bool { q.lock.Lock() - defer q.lock.Unlock() - - return q.resultSlots(q.blockPendPool, q.blockDonePool) <= 0 + t := q.resultSlots(q.blockPendPool, q.blockDonePool) <= 0 + q.lock.Unlock() + if t { + throttleBlockCounter.Inc(1) + } + return t } // ShouldThrottleReceipts checks if the download should be throttled (active receipt // fetches exceed block cache). func (q *queue) ShouldThrottleReceipts() bool { q.lock.Lock() - defer q.lock.Unlock() - - return q.resultSlots(q.receiptPendPool, q.receiptDonePool) <= 0 + t := q.resultSlots(q.receiptPendPool, q.receiptDonePool) <= 0 + q.lock.Unlock() + if t { + throttleReceiptCounter.Inc(1) + } + return t } // resultSlots calculates the number of results slots available for requests @@ -238,30 +244,19 @@ func (q *queue) ShouldThrottleReceipts() bool { // cache. func (q *queue) resultSlots(pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}) int { // Calculate the maximum length capped by the memory limit - limit := len(q.resultCache) - if common.StorageSize(len(q.resultCache))*q.resultSize > common.StorageSize(blockCacheMemory) { + cacheItems := len(q.resultCache.items) + limit := cacheItems + if common.StorageSize(cacheItems)*q.resultSize > common.StorageSize(blockCacheMemory) { limit = int((common.StorageSize(blockCacheMemory) + q.resultSize - 1) / q.resultSize) } // Calculate the number of slots already finished - finished := 0 - for _, result := range q.resultCache[:limit] { - if result == nil { - break - } - if _, ok := donePool[result.Hash]; ok { - finished++ - } - } + finished := q.resultCache.countCompleted() // Calculate the number of slots currently downloading pending := 0 + //iterations := 0 for _, request := range pendPool { - for _, header := range request.Headers { - if header.Number.Uint64() < q.resultOffset+uint64(limit) { - pending++ - } - } + pending += len(request.Headers) } - // Return the free slots to distribute return limit - finished - pending } @@ -310,6 +305,8 @@ 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 { @@ -328,18 +325,44 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header { log.Warn("Header already scheduled for block fetch", "number", header.Number, "hash", hash) continue } - if _, ok := q.receiptTaskPool[hash]; ok { - log.Warn("Header already scheduled for receipt fetch", "number", header.Number, "hash", hash) - continue - } - // Queue the header for content retrieval q.blockTaskPool[hash] = header q.blockTaskQueue.Push(header, -int64(header.Number.Uint64())) - + // Queue for receipt retrieval if q.mode == FastSync { + 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())) } + //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++ @@ -350,65 +373,49 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header { // Results retrieves and permanently removes a batch of fetch results from // the cache. the result slice will be empty if the queue has been closed. func (q *queue) Results(block bool) []*fetchResult { + + // abort early if there are no items and non-blocking requested + if !q.resultCache.HasCompletedItems() && !block { + return nil + } + q.lock.Lock() defer q.lock.Unlock() - // Count the number of items available for processing - nproc := q.countProcessableItems() - for nproc == 0 && !q.closed { + results := q.resultCache.GetCompleted(maxResultsProcess) + for len(results) == 0 && !q.closed { if !block { return nil } q.active.Wait() - nproc = q.countProcessableItems() + results = q.resultCache.GetCompleted(maxResultsProcess) } - // Since we have a batch limit, don't pull more into "dangling" memory - if nproc > maxResultsProcess { - nproc = maxResultsProcess + // Mark results as done + for _, result := range results { + hash := result.Header.Hash() + delete(q.blockDonePool, hash) + delete(q.receiptDonePool, hash) } - results := make([]*fetchResult, nproc) - copy(results, q.resultCache[:nproc]) - if len(results) > 0 { - // Mark results as done before dropping them from the cache. - for _, result := range results { - hash := result.Header.Hash() - delete(q.blockDonePool, hash) - delete(q.receiptDonePool, hash) + // Recalculate the result item weights to prevent memory exhaustion + for _, result := range results { + size := result.Header.Size() + for _, uncle := range result.Uncles { + size += uncle.Size() } - // Delete the results from the cache and clear the tail. - copy(q.resultCache, q.resultCache[nproc:]) - for i := len(q.resultCache) - nproc; i < len(q.resultCache); i++ { - q.resultCache[i] = nil + for _, receipt := range result.Receipts { + size += receipt.Size() } - // Advance the expected block number of the first cache entry. - q.resultOffset += uint64(nproc) - - // Recalculate the result item weights to prevent memory exhaustion - for _, result := range results { - size := result.Header.Size() - for _, uncle := range result.Uncles { - size += uncle.Size() - } - for _, receipt := range result.Receipts { - size += receipt.Size() - } - for _, tx := range result.Transactions { - size += tx.Size() - } - q.resultSize = common.StorageSize(blockCacheSizeWeight)*size + (1-common.StorageSize(blockCacheSizeWeight))*q.resultSize + for _, tx := range result.Transactions { + size += tx.Size() } + q.resultSize = common.StorageSize(blockCacheSizeWeight)*size + (1-common.StorageSize(blockCacheSizeWeight))*q.resultSize } return results } // countProcessableItems counts the processable items. func (q *queue) countProcessableItems() int { - for i, result := range q.resultCache { - if result == nil || result.Pending > 0 { - return i - } - } - return len(q.resultCache) + return q.resultCache.CountCompleted() } // ReserveHeaders reserves a set of headers for the given peer, skipping any @@ -455,26 +462,39 @@ func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest { // previously failed downloads. Beside the next batch of needed fetches, it also // returns a flag whether empty blocks were queued requiring processing. func (q *queue) ReserveBodies(p *peerConnection, count int) (*fetchRequest, bool, error) { - isNoop := func(header *types.Header) bool { - return header.TxHash == types.EmptyRootHash && header.UncleHash == types.EmptyUncleHash - } q.lock.Lock() defer q.lock.Unlock() - return q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, isNoop) + request, bleh, e := q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, 0x01) + + //if request != nil { + // info := []string{} + // for _, x := range request.Headers { + // info = append(info, fmt.Sprintf("%d ", x.Number)) + // } + // fmt.Printf("peer %v reserved bodies: %v\n", p.id, info) + //} + return request, bleh, e + } // ReserveReceipts reserves a set of receipt fetches for the given peer, skipping // any previously failed downloads. Beside the next batch of needed fetches, it // also returns a flag whether empty receipts were queued requiring importing. func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bool, error) { - isNoop := func(header *types.Header) bool { - return header.ReceiptHash == types.EmptyRootHash - } q.lock.Lock() defer q.lock.Unlock() - return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, isNoop) + request, bleh, e := q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, 0x02) + + //if request != nil { + // info := []string{} + // for _, x := range request.Headers { + // info = append(info, fmt.Sprintf("%d ", x.Number)) + // } + // fmt.Printf("peer %v reserved receipts: %v\n", p.id, info) + //} + return request, bleh, e } // reserveHeaders reserves a set of data download operations for a given peer, @@ -485,7 +505,7 @@ func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bo // reason the lock is not obtained in here is because the parameters already need // to access the queue, so they already need a lock anyway. func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque, - pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, isNoop func(*types.Header) bool) (*fetchRequest, bool, error) { + pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, bitFlag uint8) (*fetchRequest, bool, error) { // Short circuit if the pool has been depleted, or if the peer's already // downloading something (sanity check not to corrupt state) if taskQueue.Empty() { @@ -500,37 +520,36 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common // Retrieve a batch of tasks, skipping previously failed ones send := make([]*types.Header, 0, count) skip := make([]*types.Header, 0) - progress := false + for proc := 0; proc < space && len(send) < count && !taskQueue.Empty(); proc++ { header := taskQueue.PopItem().(*types.Header) hash := header.Hash() - - // If we're the first to request this task, initialise the result container - index := int(header.Number.Int64() - int64(q.resultOffset)) - if index >= len(q.resultCache) || index < 0 { - common.Report("index allocation went beyond available resultCache space") - return nil, false, errInvalidChain - } - if q.resultCache[index] == nil { - components := 1 - if q.mode == FastSync { - components = 2 - } - q.resultCache[index] = &fetchResult{ - Pending: components, - Hash: hash, - Header: header, - } - } - // If this fetch task is a noop, skip this fetch operation - if isNoop(header) { + stale, item, _ := q.resultCache.AddFetch(header, q.mode == FastSync) + if stale { + // Don't put back in the task queue, this item has already been + // delivered upstream + progress = true donePool[hash] = struct{}{} delete(taskPool, hash) - - space, proc = space-1, proc-1 - q.resultCache[index].Pending-- + proc-- + continue + } + if item == nil { + // There are no resultslots available. Put it back in the task queue + taskQueue.Push(header, -int64(header.Number.Uint64())) + // Set progress to true -- otherwise the peer will get + // penalized for not 'accepting' requests, while in fact it's + // because we don't have room for more results right now progress = true + break + } + // Any work to be done? + if item.Pending&bitFlag == 0 { + progress = true + donePool[hash] = struct{}{} + delete(taskPool, hash) + proc-- continue } // Otherwise unless the peer is known not to have the data, add to the retrieve list @@ -544,7 +563,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common for _, header := range skip { taskQueue.Push(header, -int64(header.Number.Uint64())) } - if progress { + if q.resultCache.HasCompletedItems() { // Wake Results, resultCache was modified q.active.Signal() } @@ -558,7 +577,6 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common Time: time.Now(), } pendPool[p.id] = request - return request, progress, nil } @@ -776,11 +794,11 @@ func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh // also wakes any threads waiting for data delivery. func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLists [][]*types.Header) (int, error) { - validate := func(index int, txHash, uncleHash, receiptHash common.Hash) error { - if types.DeriveSha(types.Transactions(txLists[index])) != txHash { + validate := func(index int, header *types.Header) error { + if types.DeriveSha(types.Transactions(txLists[index])) != header.TxHash { return errInvalidBody } - if types.CalcUncleHash(uncleLists[index]) != uncleHash { + if types.CalcUncleHash(uncleLists[index]) != header.UncleHash { return errInvalidBody } return nil @@ -789,6 +807,8 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLi reconstruct := func(index int, result *fetchResult) { result.Transactions = txLists[index] result.Uncles = uncleLists[index] + // clear body flag, AND with 1111 1110 + result.Pending &= 0xfe } return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, bodyReqTimer, len(txLists), validate, reconstruct) } @@ -798,14 +818,16 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLi // and also wakes any threads waiting for data delivery. func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int, error) { - validate := func(index int, txHash, uncleHash, receiptHash common.Hash) error { - if types.DeriveSha(types.Receipts(receiptList[index])) != receiptHash { + validate := func(index int, header *types.Header) error { + if types.DeriveSha(types.Receipts(receiptList[index])) != header.ReceiptHash { return errInvalidReceipt } return nil } reconstruct := func(index int, result *fetchResult) { result.Receipts = receiptList[index] + // clear bit 1, AND with 1111 1101 + result.Pending &= 0xfd } return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, receiptReqTimer, len(receiptList), validate, reconstruct) } @@ -815,7 +837,7 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int, // This method obtains the lock as needed func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque, pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, reqTimer metrics.Timer, - results int, validate func(index int, txHash, uncleHash, receiptHash common.Hash) error, reconstruct func(index int, result *fetchResult)) (int, error) { + results int, validate func(index int, header *types.Header) error, reconstruct func(index int, result *fetchResult)) (int, error) { q.lock.Lock() // Short circuit if the data was never requested @@ -840,41 +862,42 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQ failure error i int ) - // Need the read lock to access resultcache - q.lock.RLock() for _, header := range request.Headers { // Short circuit assembly if no more fetch results are found if i >= results { break } // Validate the fields - if err := validate(i, header.TxHash, header.UncleHash, header.ReceiptHash); err != nil { + if err := validate(i, header); err != nil { failure = err break } header.Hash() i++ } - q.lock.RUnlock() q.lock.Lock() var acceptCount = 0 for _, header := range request.Headers[:i] { - index := int(header.Number.Int64() - int64(q.resultOffset)) - if index >= len(q.resultCache) || index < 0 { - // TODO! this should probably be errStaleDelivery instead - failure = errStaleDelivery - break - } - if res := q.resultCache[index]; res != nil { - hash := header.Hash() - donePool[hash] = struct{}{} - reconstruct(acceptCount, res) - res.Pending-- - delete(taskPool, hash) - } - // else: betweeen here and above, some other peer filled this result - // we just ignore and move on + // TODO @holiman + // q.resultCache.deliver(header, data, reconstruct) + // or + // q.resultCache.deliverBody , q.resultCache.deliverReceipts + // or + // q.resultCache.deliverBodies( bodies, headers[0]) + if res, stale, err := q.resultCache.GetFetchResult(header); err == nil { + reconstruct(acceptCount, res) + } else { + // else: betweeen here and above, some other peer filled this result + // we just ignore and move on + // TODO @holiman + // figure out how this can happen (it shouldn't) + log.Info("delivery stale?", "err", err, "stale", stale) + failure = errStaleDelivery + } + hash := header.Hash() + donePool[hash] = struct{}{} + delete(taskPool, hash) // Clean up a successful fetch request.Headers[acceptCount] = nil acceptCount++ @@ -909,8 +932,6 @@ func (q *queue) Prepare(offset uint64, mode SyncMode) { defer q.lock.Unlock() // Prepare the queue for sync results - if q.resultOffset < offset { - q.resultOffset = offset - } + q.resultCache.Prepare(offset) q.mode = mode } diff --git a/eth/downloader/resultcache.go b/eth/downloader/resultcache.go new file mode 100644 index 0000000000..8f0ae8f898 --- /dev/null +++ b/eth/downloader/resultcache.go @@ -0,0 +1,258 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library 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 Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// resultcache implements a structure for maintaining fetchResults, tracking their +// download-progress and delivering (finished) results + +package downloader + +import ( + "fmt" + "github.com/ethereum/go-ethereum/log" + "sync" + "sync/atomic" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" +) + +type resultStore struct { + items []*fetchResult // Downloaded but not yet delivered fetch results + lock *sync.RWMutex // lock protect internals + resultOffset uint64 // Offset of the first cached fetch result in the block chain + resultSize common.StorageSize // Approximate size of a block (exponential moving average) + + // Internal index of first non-completed entry, updated atomically when needed. + // If all items are complete, this will equal length(items), so + // *important* : is not safe to use for indexing without checking against length + indexIncomplete int32 +} + +func newResultStore(size int) *resultStore { + return &resultStore{ + resultOffset: 0, + items: make([]*fetchResult, size), + resultSize: 0, // TODO: use a saner default, left at zero as it was legacy + lock: new(sync.RWMutex), + } +} + +// AddFetch adds a header for body/receipt fetching. +// returning +// stale -- if true, this item is already passed, and should not be requested again +// fetchResult -- if `nil`, that means no fetch was created, and that the +// if an error is returned, that most likely means backpressure prevents the results from expanding, +// and someone needs to take care of results +func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale bool, item *fetchResult, err error) { + hash := header.Hash() + r.lock.RLock() + item, _, stale, err = r.getFetchResult(header) + if err != nil { + r.lock.RUnlock() + log.Info("resultcache addfetch err [1]", "error", err.Error()) + // can't create a fetchResult right away + return stale, nil, err + } + if item != nil { + r.lock.RUnlock() + return false, item, nil + } + r.lock.RUnlock() + // Need to create a fetchresult, and as we've just release the Rlock, + // we need to check again after obtaining the writelock + r.lock.Lock() + defer r.lock.Unlock() + var index int + item, index, stale, err = r.getFetchResult(header) + if err != nil { + // can't create a fetchResult right away + log.Info("resultcache addfetch err [2]", "error", err.Error()) + return stale, nil, err + + } + if item == nil { + item = &fetchResult{ + Hash: hash, + Header: header, + } + // Need to fetch body? + if !header.EmptyBody() { + // yes + item.Pending |= 0x1 + } + // Do we need to fetch receipts? + if fastSync && !header.EmptyReceipts() { + item.Pending |= 0x2 + } + r.items[index] = item + } + return false, item, nil + +} + +func (r *resultStore) GetFetchResult(header *types.Header) (*fetchResult, bool, error) { + r.lock.RLock() + defer r.lock.RUnlock() + res, _, stale, err := r.getFetchResult(header) + return res, stale, err +} + +// getFetchResult returns the fetchResult corresponding to the given item, and the index where +// the result is stored. +// There are two ways it can error: +// 1. The header is too far off in the future, and we don't have room for it. +// 2. The header is stale, and the results for that header has already been delivered upstream. +func (r *resultStore) getFetchResult(header *types.Header) (item *fetchResult, index int, stale bool, err error) { + index = int(header.Number.Int64() - int64(r.resultOffset)) + if index >= len(r.items) { + err = fmt.Errorf("index allocation went beyond available resultStore space "+ + "(index [%d] = header [%d] - resultOffset [%d], len(resultStore) = %d", + index, header.Number.Int64(), r.resultOffset, len(r.items)) + return item, index, stale, err + } + if index < 0 { + stale = true + err = fmt.Errorf("index allocation went beyond available resultStore space "+ + "(index [%d] = header [%d] - resultOffset [%d], len(resultStore) = %d", + index, header.Number.Int64(), r.resultOffset, len(r.items)) + return item, index, stale, err + } + item = r.items[index] + return item, index, stale, err +} + +// numberSpan returns the header number start and end, for the headers +// currently "allocated" for download (both completed, in-flight and pending) +func (r *resultStore) NumberSpan() (uint64, uint64) { + r.lock.RLock() + defer r.lock.RUnlock() + return r.resultOffset, r.resultOffset + uint64(len(r.items)) + +} + +// hasCompletedItems returns true if there are processable items available +// this method is cheaper than countCompleted +func (r *resultStore) HasCompletedItems() bool { + r.lock.RLock() + defer r.lock.RUnlock() + if len(r.items) == 0 { + return false + } + if item := r.items[0]; item != nil && item.Pending == 0 { + return true + } + return false +} + +// CountCompleted returns the number of items completed +func (r *resultStore) CountCompleted() int { + r.lock.RLock() + defer r.lock.RUnlock() + return r.countCompleted() +} + +// countCompleted returns the number of items completed +// assumes (at least) rlock is held +func (r *resultStore) countCompleted() int { + // We iterate from the already known complete point, and see + // if any more has completed since last count + // debug + /* + var ( + nils = 0 + fins = 0 + bodyneeds = 0 + receiptneeds = 0 + ) + var ctx []interface{} + for _, item := range r.items { + if item == nil { + nils++ + } else { + if item.Pending == 0 { + fins++ + } else { + if item.Pending&0x01 != 0 { + bodyneeds++ + } else { + receiptneeds++ + } + } + } + } + ctx = append(ctx, "items", len(r.items), "nils", nil, "fins", fins, + "needB", bodyneeds, "needR", receiptneeds) + */ + /// end debug + index := atomic.LoadInt32(&r.indexIncomplete) + for ; ; index++ { + if index >= int32(len(r.items)) { + break + } + result := r.items[index] + if result == nil || result.Pending > 0 { + break + } + } + /* + if index < int32(len(r.items)) { + //ctx = append(ctx, []interface{}{"index", index, "blocknum", uint64(index) + r.resultOffset}...) + if r.items[index] != nil { + log.Info("resultstore", ctx...) + } else { + ctx = append(ctx, []interface{}{"first missing", "nil"}...) + log.Info("resultstore", ctx...) + } + } else { + ctx = append(ctx, []interface{}{"first missing", "out of range"}...) + log.Info("resultstore", ctx...) + } + */ + atomic.StoreInt32(&r.indexIncomplete, index) + return int(index) +} + +// getCompleted returns the next batch of completed fetchresults +func (r *resultStore) GetCompleted(limit int) []*fetchResult { + r.lock.Lock() + defer r.lock.Unlock() + + completed := r.countCompleted() + if limit > completed { + limit = completed + } + results := make([]*fetchResult, limit) + copy(results, r.items[:limit]) + + // Delete the results from the cache and clear the tail. + copy(r.items, r.items[limit:]) + for i := len(r.items) - limit; i < len(r.items); i++ { + r.items[i] = nil + } + // Advance the expected block number of the first cache entry. + r.resultOffset += uint64(limit) + // And subtract the number of items from our two indexes + atomic.StoreInt32(&r.indexIncomplete, int32(completed-limit)) + return results +} + +func (r *resultStore) Prepare(offset uint64) { + r.lock.Lock() + if r.resultOffset < offset { + r.resultOffset = offset + } + r.lock.Unlock() +} diff --git a/eth/fetcher/block_fetcher.go b/eth/fetcher/block_fetcher.go index 7690a53862..b7aa47e5a1 100644 --- a/eth/fetcher/block_fetcher.go +++ b/eth/fetcher/block_fetcher.go @@ -538,40 +538,51 @@ func (f *BlockFetcher) loop() { return } bodyFilterInMeter.Mark(int64(len(task.transactions))) - blocks := []*types.Block{} - for i := 0; i < len(task.transactions) && i < len(task.uncles); i++ { - // Match up a body to any possible completion request - matched := false - - for hash, announce := range f.completing { - if f.queued[hash] == nil { - txnHash := types.DeriveSha(types.Transactions(task.transactions[i])) - uncleHash := types.CalcUncleHash(task.uncles[i]) - - if txnHash == announce.header.TxHash && uncleHash == announce.header.UncleHash && announce.origin == task.peer { - // Mark the body matched, reassemble if still unknown - matched = true - - if f.getBlock(hash) == nil { - block := types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i]) - block.ReceivedAt = task.time - - blocks = append(blocks, block) - } else { - f.forgetHash(hash) - } + // abort early if there's nothing explicitly requested + if len(f.completing) > 0 { + for i := 0; i < len(task.transactions) && i < len(task.uncles); i++ { + // Match up a body to any possible completion request + var ( + matched = false + uncleHash common.Hash // calculated lazily and reused + txnHash common.Hash // calculated lazily and reused + ) + for hash, announce := range f.completing { + if f.queued[hash] != nil || announce.origin != task.peer { + continue } + if uncleHash == (common.Hash{}) { + uncleHash = types.CalcUncleHash(task.uncles[i]) + } + if uncleHash != announce.header.UncleHash { + continue + } + if txnHash == (common.Hash{}) { + txnHash = types.DeriveSha(types.Transactions(task.transactions[i])) + } + if txnHash != announce.header.TxHash { + continue + } + // Mark the body matched, reassemble if still unknown + matched = true + if f.getBlock(hash) == nil { + block := types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i]) + block.ReceivedAt = task.time + blocks = append(blocks, block) + } else { + f.forgetHash(hash) + } + + } + if matched { + task.transactions = append(task.transactions[:i], task.transactions[i+1:]...) + task.uncles = append(task.uncles[:i], task.uncles[i+1:]...) + i-- + continue } } - if matched { - task.transactions = append(task.transactions[:i], task.transactions[i+1:]...) - task.uncles = append(task.uncles[:i], task.uncles[i+1:]...) - i-- - continue - } } - bodyFilterOutMeter.Mark(int64(len(task.transactions))) select { case filter <- task: