diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index cd5395c9c5..25830239ef 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -1090,10 +1090,9 @@ func (d *Downloader) fillHeaderSkeleton(from uint64, skeleton []*types.Header) ( pack := packet.(*headerPack) return d.queue.DeliverHeaders(pack.peerID, pack.headers, d.headerProcCh) } - expire = func() map[string]int { return d.queue.ExpireHeaders(d.requestTTL()) } - throttle = func() bool { return false } - reserve = func(p *peerConnection, count int) (*fetchRequest, bool, error) { - return d.queue.ReserveHeaders(p, count), false, nil + expire = func() map[string]int { return d.queue.ExpireHeaders(d.requestTTL()) } + reserve = func(p *peerConnection, count int) (*fetchRequest, bool, bool, error) { + return d.queue.ReserveHeaders(p, count), false, false, nil } fetch = func(p *peerConnection, req *fetchRequest) error { return p.FetchHeaders(req.From, MaxHeaderFetch) } capacity = func(p *peerConnection) int { return p.HeaderCapacity(d.requestRTT()) } @@ -1102,7 +1101,7 @@ func (d *Downloader) fillHeaderSkeleton(from uint64, skeleton []*types.Header) ( } ) err := d.fetchParts(d.headerCh, deliver, d.queue.headerContCh, expire, - d.queue.PendingHeaders, d.queue.InFlightHeaders, throttle, reserve, + d.queue.PendingHeaders, d.queue.InFlightHeaders, reserve, nil, fetch, d.queue.CancelHeaders, capacity, d.peers.HeaderIdlePeers, setIdle, "headers") log.Debug("Skeleton fill terminated", "err", err) @@ -1128,7 +1127,7 @@ func (d *Downloader) fetchBodies(from uint64) error { setIdle = func(p *peerConnection, accepted int, deliveryTime time.Time) { p.SetBodiesIdle(accepted, deliveryTime) } ) err := d.fetchParts(d.bodyCh, deliver, d.bodyWakeCh, expire, - d.queue.PendingBlocks, d.queue.InFlightBlocks, d.queue.ShouldThrottleBlocks, d.queue.ReserveBodies, + d.queue.PendingBlocks, d.queue.InFlightBlocks, d.queue.ReserveBodies, d.bodyFetchHook, fetch, d.queue.CancelBodies, capacity, d.peers.BodyIdlePeers, setIdle, "bodies") log.Debug("Block body download terminated", "err", err) @@ -1154,7 +1153,7 @@ func (d *Downloader) fetchReceipts(from uint64) error { } ) err := d.fetchParts(d.receiptCh, deliver, d.receiptWakeCh, expire, - d.queue.PendingReceipts, d.queue.InFlightReceipts, d.queue.ShouldThrottleReceipts, d.queue.ReserveReceipts, + d.queue.PendingReceipts, d.queue.InFlightReceipts, d.queue.ReserveReceipts, d.receiptFetchHook, fetch, d.queue.CancelReceipts, capacity, d.peers.ReceiptIdlePeers, setIdle, "receipts") log.Debug("Transaction receipt download terminated", "err", err) @@ -1187,7 +1186,7 @@ func (d *Downloader) fetchReceipts(from uint64) error { // - setIdle: network callback to set a peer back to idle and update its estimated capacity (traffic shaping) // - kind: textual label of the type being downloaded to display in log mesages func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack) (int, error), wakeCh chan bool, - expire func() map[string]int, pending func() int, inFlight func() bool, throttle func() bool, reserve func(*peerConnection, int) (*fetchRequest, bool, error), + expire func() map[string]int, pending func() int, inFlight func() bool, reserve func(*peerConnection, int) (*fetchRequest, bool, bool, error), fetchHook func([]*types.Header), fetch func(*peerConnection, *fetchRequest) error, cancel func(*fetchRequest), capacity func(*peerConnection) int, idle func() ([]*peerConnection, int), setIdle func(*peerConnection, int, time.Time), kind string) error { @@ -1306,56 +1305,55 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack) // Send a download request to all idle peers, until throttled progressed, throttled, running := false, false, inFlight() idles, total := idle() - + pendCount := 1 for _, peer := range idles { // Short circuit if throttling activated - if throttle() { - throttled = true + if throttled { break } // Short circuit if there is no more available task. - if pending() == 0 { + if pendCount = pending(); pendCount == 0 { break } // Reserve a chunk of fetches for a peer. A nil can mean either that // no more headers are available, or that the peer is known not to // have them. - request, progress, err := reserve(peer, capacity(peer)) + request, progress, throttle, err := reserve(peer, capacity(peer)) if err != nil { + log.Info("Error in loop", "err", err) return err } if progress { progressed = true } - if request == nil { - // This means we've over-committed, and reserving returns nil - // because there's no space for the top-prio header to download. - // No need to continue this loop + if throttle { throttled = true - break + throttleBlockCounter.Inc(1) } - if request.From > 0 { - peer.log.Trace("Requesting new batch of data", "type", kind, "from", request.From) - } else { - peer.log.Trace("Requesting new batch of data", "type", kind, "count", len(request.Headers), "from", request.Headers[0].Number) - } - // Fetch the chunk and make sure any errors return the hashes to the queue - if fetchHook != nil { - fetchHook(request.Headers) - } - if err := fetch(peer, request); err != nil { - // Although we could try and make an attempt to fix this, this error really - // means that we've double allocated a fetch task to a peer. If that is the - // case, the internal state of the downloader and the queue is very wrong so - // better hard crash and note the error instead of silently accumulating into - // a much bigger issue. - panic(fmt.Sprintf("%v: %s fetch assignment failed", peer, kind)) + if request != nil { + if request.From > 0 { + peer.log.Trace("Requesting new batch of data", "type", kind, "from", request.From) + } else { + peer.log.Trace("Requesting new batch of data", "type", kind, "count", len(request.Headers), "from", request.Headers[0].Number) + } + // Fetch the chunk and make sure any errors return the hashes to the queue + if fetchHook != nil { + fetchHook(request.Headers) + } + if err := fetch(peer, request); err != nil { + // Although we could try and make an attempt to fix this, this error really + // means that we've double allocated a fetch task to a peer. If that is the + // case, the internal state of the downloader and the queue is very wrong so + // better hard crash and note the error instead of silently accumulating into + // a much bigger issue. + panic(fmt.Sprintf("%v: %s fetch assignment failed", peer, kind)) + } } running = true } // Make sure that we have peers available for fetching. If all peers have been tried // and all failed throw an error - if !progressed && !throttled && !running && len(idles) == total && pending() > 0 { + if !progressed && !throttled && !running && len(idles) == total && pendCount > 0 { return errPeersUnavailable } } @@ -1482,7 +1480,7 @@ 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) + log.Info("Invalid header encountered", "number", chunk[n].Number, "hash", chunk[n].Hash(), "err", err, "parent", chunk[n].ParentHash) return errInvalidChain } // All verifications passed, store newly found uncertain headers @@ -1637,6 +1635,7 @@ func (d *Downloader) processFastSyncContent(latest *types.Header) error { } } P, beforeP, afterP := splitAroundPivot(pivot, results) + results = nil if err := d.commitFastSyncData(beforeP, sync); err != nil { return err } @@ -1674,6 +1673,15 @@ func (d *Downloader) processFastSyncContent(latest *types.Header) error { } func splitAroundPivot(pivot uint64, results []*fetchResult) (p *fetchResult, before, after []*fetchResult) { + if len(results) == 0 { + return + } + if lastNum := results[len(results)-1].Header.Number.Uint64(); lastNum < pivot { + // the pivot is somewhere in the future + before = results + return + } + // This can also be optimized, but only happens very seldom for _, result := range results { num := result.Header.Number.Uint64() switch { @@ -1704,7 +1712,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.Info("Inserting fast-sync blocks", "items", len(results), + log.Debug("Inserting fast-sync blocks", "items", len(results), "firstnum", first.Number, "firsthash", first.Hash(), "lastnumn", last.Number, "lasthash", last.Hash(), ) diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index 32a96ba412..f0dfff5283 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -267,7 +267,7 @@ func (p *peerConnection) setIdle(elapsed time.Duration, delivered int, throughpu return } // Otherwise update the throughput with a new measurement - if elapsed <= 0{ + if elapsed <= 0 { elapsed = 1 // +1 (ns) to ensure non-zero divisor } measured := float64(delivered) / (float64(elapsed) / float64(time.Second)) diff --git a/eth/downloader/peer_test.go b/eth/downloader/peer_test.go index 082cf86610..2bb6291887 100644 --- a/eth/downloader/peer_test.go +++ b/eth/downloader/peer_test.go @@ -5,33 +5,33 @@ import ( "testing" ) -func TestPeerThroughputSorting(t *testing.T){ +func TestPeerThroughputSorting(t *testing.T) { a := &peerConnection{ - id:"a", - headerThroughput:1.25, + id: "a", + headerThroughput: 1.25, } b := &peerConnection{ - id: "b", - headerThroughput:1.21, + id: "b", + headerThroughput: 1.21, } c := &peerConnection{ - id: "c", - headerThroughput:1.23, + id: "c", + headerThroughput: 1.23, } - peers := []*peerConnection{a,b,c} + peers := []*peerConnection{a, b, c} tps := []float64{a.headerThroughput, - b.headerThroughput, c.headerThroughput} + b.headerThroughput, c.headerThroughput} sortPeers := &peerThroughputSort{peers, tps} sort.Sort(sortPeers) - if got, exp := sortPeers.p[0].id , "a"; got != exp{ + 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{ + 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{ + 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 1058285c43..390900711b 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -23,6 +23,7 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "time" "github.com/ethereum/go-ethereum/common" @@ -32,6 +33,11 @@ import ( "github.com/ethereum/go-ethereum/metrics" ) +const ( + BodyType = 0 + ReceiptType = 1 +) + var ( blockCacheItems = 8192 // Maximum number of blocks to cache before throttling the download blockCacheMemory = 64 * 1024 * 1024 // Maximum amount of memory to use for block caching @@ -54,7 +60,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 uint8 // Number of data fetches still pending + pending int32 // Flag telling what deliveries are outstanding Hash common.Hash // Hash of the header to prevent recalculating Header *types.Header @@ -63,6 +69,52 @@ type fetchResult struct { Receipts types.Receipts } +func newFetchResult(header *types.Header, fastSync bool) *fetchResult { + item := &fetchResult{ + Hash: header.Hash(), + Header: header, + } + if !header.EmptyBody() { + item.pending = 1 + } + if fastSync && !header.EmptyReceipts() { + item.pending += 2 + } + return item +} + +// SetBodyDone flags the body as finished. +func (f *fetchResult) SetBodyDone() { + if v := atomic.LoadInt32(&f.pending); v == 1 || v == 3 { + atomic.AddInt32(&f.pending, -1) + } +} + +// AllDone checks 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 || v == 3 { + atomic.AddInt32(&f.pending, -2) + } +} + +// CheckDone checks if the given type is done already +func (f *fetchResult) Done(typ int) bool { + v := atomic.LoadInt32(&f.pending) + switch typ { + case BodyType: + return !(v == 1 || v == 3) + case ReceiptType: + return !(v == 2 || v == 3) + default: + return false + } +} + // queue represents hashes that are either need fetching or are being fetched type queue struct { mode SyncMode // Synchronisation mode to decide on the block parts to schedule for fetching @@ -82,12 +134,10 @@ type queue struct { blockTaskPool map[common.Hash]*types.Header // [eth/62] Pending block (body) retrieval tasks, mapping hashes to headers blockTaskQueue *prque.Prque // [eth/62] Priority queue of the headers to fetch the blocks (bodies) for blockPendPool map[string]*fetchRequest // [eth/62] Currently pending block (body) retrieval operations - blockDonePool map[common.Hash]struct{} // [eth/62] Set of the completed block (body) fetches receiptTaskPool map[common.Hash]*types.Header // [eth/63] Pending receipt retrieval tasks, mapping hashes to headers 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 - receiptDonePool map[common.Hash]struct{} // [eth/63] Set of the completed receipt fetches resultCache *resultStore // Downloaded but not yet delivered fetch results //resultOffset uint64 // Offset of the first cached fetch result in the block chain @@ -107,15 +157,12 @@ func newQueue(blockCacheLimit int) *queue { blockTaskPool: make(map[common.Hash]*types.Header), blockTaskQueue: prque.New(nil), blockPendPool: make(map[string]*fetchRequest), - blockDonePool: make(map[common.Hash]struct{}), receiptTaskPool: make(map[common.Hash]*types.Header), receiptTaskQueue: prque.New(nil), receiptPendPool: make(map[string]*fetchRequest), - receiptDonePool: make(map[common.Hash]struct{}), - //resultCache: make([]*fetchResult, blockCacheItems), - resultCache: newResultStore(blockCacheLimit), - active: sync.NewCond(lock), - lock: lock, + resultCache: newResultStore(blockCacheLimit * 2), + active: sync.NewCond(lock), + lock: lock, } } @@ -133,14 +180,12 @@ func (q *queue) Reset() { q.blockTaskPool = make(map[common.Hash]*types.Header) q.blockTaskQueue.Reset() q.blockPendPool = make(map[string]*fetchRequest) - q.blockDonePool = make(map[common.Hash]struct{}) q.receiptTaskPool = make(map[common.Hash]*types.Header) q.receiptTaskQueue.Reset() q.receiptPendPool = make(map[string]*fetchRequest) - q.receiptDonePool = make(map[common.Hash]struct{}) - q.resultCache = newResultStore(blockCacheItems) + q.resultCache = newResultStore(blockCacheItems * 2) } // Close marks the end of the sync, unblocking Results. @@ -210,54 +255,8 @@ func (q *queue) Idle() bool { queued := q.blockTaskQueue.Size() + q.receiptTaskQueue.Size() pending := len(q.blockPendPool) + len(q.receiptPendPool) - cached := len(q.blockDonePool) + len(q.receiptDonePool) - return (queued + pending + cached) == 0 -} - -// ShouldThrottleBlocks checks if the download should be throttled (active block (body) -// fetches exceed block cache). -func (q *queue) ShouldThrottleBlocks() bool { - q.lock.Lock() - 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() - 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 -// whilst adhering to both the item and the memory limit too of the results -// cache. -func (q *queue) resultSlots(pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}) int { - // Calculate the maximum length capped by the memory limit - 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 := q.resultCache.countCompleted() - // Calculate the number of slots currently downloading - pending := 0 - //iterations := 0 - for _, request := range pendPool { - pending += len(request.Headers) - } - return limit - finished - pending + return (queued + pending) == 0 } // ScheduleSkeleton adds a batch of header retrieval tasks to the queue to fill @@ -372,32 +371,26 @@ 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. +// This is 'thread-safe', but assumes that there are not two simultaneous +// callers to Results (both will modify q.resultSize) 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() - results := q.resultCache.GetCompleted(maxResultsProcess) for len(results) == 0 && !q.closed { if !block { return nil } + q.lock.Lock() q.active.Wait() + q.lock.Unlock() results = q.resultCache.GetCompleted(maxResultsProcess) } - // Mark results as done - 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 { + // Recalculate the result item weights to prevent memory exhaustion size := result.Header.Size() for _, uncle := range result.Uncles { size += uncle.Size() @@ -408,14 +401,33 @@ func (q *queue) Results(block bool) []*fetchResult { for _, tx := range result.Transactions { size += tx.Size() } - q.resultSize = common.StorageSize(blockCacheSizeWeight)*size + (1-common.StorageSize(blockCacheSizeWeight))*q.resultSize + q.resultSize = common.StorageSize(blockCacheSizeWeight)*size + + (1-common.StorageSize(blockCacheSizeWeight))*q.resultSize + } + // Using the newly calibrated resultsize, figure out the new throttle limit + // on the result cache + throttleThreshold := uint64((common.StorageSize(blockCacheMemory) + q.resultSize - 1) / q.resultSize) + q.resultCache.SetThrottleThreshold(throttleThreshold) + // log some info at certain times + if time.Now().Second()&0xa == 0 { + info := q.Stats() + info = append(info, "throttle", throttleThreshold) + log.Info("queue stats", info...) } return results } -// countProcessableItems counts the processable items. -func (q *queue) countProcessableItems() int { - return q.resultCache.CountCompleted() +func (q *queue) Stats() []interface{} { + q.lock.RLock() + defer q.lock.RUnlock() + return q.stats() +} +func (q *queue) stats() []interface{} { + return []interface{}{ + "receiptTaskQueue", q.receiptTaskQueue.Size(), + "blockTaskQueue", q.blockTaskQueue.Size(), + "est resultSize", q.resultSize, + } } // ReserveHeaders reserves a set of headers for the given peer, skipping any @@ -461,40 +473,21 @@ func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest { // ReserveBodies reserves a set of body fetches for the given peer, skipping any // 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) { +func (q *queue) ReserveBodies(p *peerConnection, count int) (*fetchRequest, bool, bool, error) { q.lock.Lock() defer q.lock.Unlock() - 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 - + return q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, BodyType) } // 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) { +func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bool, bool, error) { q.lock.Lock() defer q.lock.Unlock() - 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 + return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, ReceiptType) } // reserveHeaders reserves a set of data download operations for a given peer, @@ -504,52 +497,68 @@ func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bo // Note, this method expects the queue lock to be already held for writing. The // 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. +// returns: +// item - the fetchRequest +// progress, bool - whether any progress was made +// throttle, bool - if the caller should throttle for a while +// error - any error that occcurred 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{}, bitFlag uint8) (*fetchRequest, bool, error) { + pendPool map[string]*fetchRequest, typ int) (*fetchRequest, bool, 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() { - return nil, false, nil + return nil, false, true, nil } if _, ok := pendPool[p.id]; ok { - return nil, false, nil + return nil, false, false, nil } - // Calculate an upper limit on the items we might fetch (i.e. throttling) - space := q.resultSlots(pendPool, donePool) - // 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++ { + throttled := false + for proc := 0; len(send) < count && !taskQueue.Empty(); proc++ { + // the task queue will pop items in order, so the highest prio block + // is also the lowest block number. header := taskQueue.PopItem().(*types.Header) + // we can ask the resultcache if this header is within the + // "prioritized" segment of blocks. If it is not, we need to throttle + hash := header.Hash() - stale, item, _ := q.resultCache.AddFetch(header, q.mode == FastSync) + stale, throttle, item, err := 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) - proc-- + proc = proc - 1 continue } - if item == nil { + if throttle { // 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 + // However, if there are any left as 'skipped', we should not tell + // the caller to throttle, since we still want some other + // peer to fetch those for us + throttled = len(skip) == 0 break } - // Any work to be done? - if item.Pending&bitFlag == 0 { - progress = true - donePool[hash] = struct{}{} + if err != nil { + // this most definitely should _not_ happen + log.Warn("reserve headers error", "error", err) + // There are no resultslots available. Put it back in the task queue + taskQueue.Push(header, -int64(header.Number.Uint64())) + break + } + if item.Done(typ) { + // If it's a noop, we can skip this task delete(taskPool, hash) - proc-- + proc = proc - 1 + progress = true continue } // Otherwise unless the peer is known not to have the data, add to the retrieve list @@ -569,7 +578,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common } // Assemble and return the block download request if len(send) == 0 { - return nil, progress, nil + return nil, progress, throttled, nil } request := &fetchRequest{ Peer: p, @@ -577,7 +586,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common Time: time.Now(), } pendPool[p.id] = request - return request, progress, nil + return request, progress, throttled, nil } // CancelHeaders aborts a fetch request, returning all pending skeleton indexes to the queue. @@ -807,10 +816,10 @@ 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 + result.SetBodyDone() } - return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, bodyReqTimer, len(txLists), validate, reconstruct) + return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, + bodyReqTimer, len(txLists), validate, reconstruct) } // DeliverReceipts injects a receipt retrieval response into the results queue. @@ -826,18 +835,19 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int, } reconstruct := func(index int, result *fetchResult) { result.Receipts = receiptList[index] - // clear bit 1, AND with 1111 1101 - result.Pending &= 0xfd + result.SetReceiptsDone() } - return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, receiptReqTimer, len(receiptList), validate, reconstruct) + return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, + receiptReqTimer, len(receiptList), validate, reconstruct) } // deliver injects a data retrieval response into the results queue. // // 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, header *types.Header) error, reconstruct func(index int, result *fetchResult)) (int, error) { +func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, + taskQueue *prque.Prque, pendPool map[string]*fetchRequest, reqTimer metrics.Timer, + 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 @@ -867,37 +877,27 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQ if i >= results { break } + header.Hash() // Validate the fields if err := validate(i, header); err != nil { failure = err break } - header.Hash() i++ } q.lock.Lock() var acceptCount = 0 for _, header := range request.Headers[:i] { - // 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) + // else: betweeen here and above, some other peer filled this result, + // or it was indeed a no-op. This should not happen, but if it does it's + // not something to panic about log.Info("delivery stale?", "err", err, "stale", stale) failure = errStaleDelivery } - hash := header.Hash() - donePool[hash] = struct{}{} - delete(taskPool, hash) + delete(taskPool, header.Hash()) // Clean up a successful fetch request.Headers[acceptCount] = nil acceptCount++ diff --git a/eth/downloader/resultcache.go b/eth/downloader/resultcache.go index 8f0ae8f898..e76a52aed8 100644 --- a/eth/downloader/resultcache.go +++ b/eth/downloader/resultcache.go @@ -21,126 +21,129 @@ 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" + "github.com/ethereum/go-ethereum/log" ) 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) + 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 // 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 + indexIncomplete int32 // atomic access + + // throttleThreshold is the limit up to which we _want_ to fill the + // results. If blocks are large, we want to limit the results to less + // than the number of available slots, and maybe only fill 1024 out of + // 8192 possible places. The queue will, at certain times, recalibrate + // this index. + throttleThreshold uint64 } 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), + resultOffset: 0, + items: make([]*fetchResult, size), + lock: new(sync.RWMutex), + throttleThreshold: 3 * uint64(size) / 4, // 75% } } -// 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() +func (r *resultStore) SetThrottleThreshold(threshold uint64) { + r.lock.Lock() + defer r.lock.Unlock() + limit := uint64(len(r.items)) * 3 / 4 + if threshold >= limit { + threshold = limit + } + r.throttleThreshold = threshold +} + +// AddFetch adds a header for body/receipt fetching. This is used when the queue +// wants to reserve headers for fetching. +// It returns the following: +// stale -- if true, this item is already passed, and should not be requested again. +// throttled -- if true, the resultcache is at capacity, and this particular header is not +// prio right now +// fetchResult -- the result to store data into +// err -- any error that occurred +func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale, throttled bool, item *fetchResult, err error) { + header.Hash() r.lock.RLock() - item, _, stale, err = r.getFetchResult(header) - if err != nil { + var index int + if item, index, stale, throttled, err = r.getFetchResult(header); 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 + return + } + if stale { + r.lock.RUnlock() + return + } + if throttled { + // Index is above the current threshold of 'prioritized' blocks, + log.Debug("resultcache throttle", "index", index, "threshold", r.throttleThreshold) + r.lock.RUnlock() + return } if item != nil { + // All good, item already exists (perhaps a receipt fetch following + // a body fetch) r.lock.RUnlock() - return false, item, nil + return } 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 - + // Same checks as above, now with wlock + if item, index, stale, throttled, err = r.getFetchResult(header); err != nil { + return + } + if stale || throttled { + return } 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 - } + item = newFetchResult(header, fastSync) r.items[index] = item } - return false, item, nil - + return } +// GetFetchResult returns the fetchResult for the given header. If the 'stale' flag +// is true, that means the header has already been delivered 'upstream'. func (r *resultStore) GetFetchResult(header *types.Header) (*fetchResult, bool, error) { r.lock.RLock() defer r.lock.RUnlock() - res, _, stale, err := r.getFetchResult(header) + 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) { +func (r *resultStore) getFetchResult(header *types.Header) (item *fetchResult, index int, stale, throttle bool, err error) { + index = int(header.Number.Int64() - int64(r.resultOffset)) + throttle = index >= int(r.throttleThreshold) + stale = index < 0 + 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 + return } - 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 + if stale { + return } 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)) - + return } // hasCompletedItems returns true if there are processable items available @@ -151,81 +154,33 @@ func (r *resultStore) HasCompletedItems() bool { if len(r.items) == 0 { return false } - if item := r.items[0]; item != nil && item.Pending == 0 { + if item := r.items[0]; item != nil && item.AllDone() { 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 +// countCompleted returns the number of items ready for delivery, stopping at +// the first non-complete item. +// It 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 { + if result == nil || !result.AllDone() { 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 +// GetCompleted returns the next batch of completed fetchResults func (r *resultStore) GetCompleted(limit int) []*fetchResult { r.lock.Lock() defer r.lock.Unlock() @@ -244,11 +199,12 @@ func (r *resultStore) GetCompleted(limit int) []*fetchResult { } // 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)) + // And subtract the number of items from our index + atomic.AddInt32(&r.indexIncomplete, int32(-limit)) return results } +// Prepare initialises the offset with the given block number func (r *resultStore) Prepare(offset uint64) { r.lock.Lock() if r.resultOffset < offset {