diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index ea18b07e88..9bbafd083b 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -367,7 +367,7 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode d.stateBloom.Close() } // Reset the queue, peer set and wake channels to clean any internal leftover state - d.queue.Reset() + d.queue.Reset(blockCacheItems) d.peers.Reset() for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} { @@ -1644,7 +1644,6 @@ 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 } diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 8fa0e97bb0..0fa8336dfa 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -83,7 +83,7 @@ func newFetchResult(header *types.Header, fastSync bool) *fetchResult { // SetBodyDone flags the body as finished. func (f *fetchResult) SetBodyDone() { - if v := atomic.LoadInt32(&f.pending); v == 1 || v == 3 { + if v := atomic.LoadInt32(&f.pending); (v & 1) != 0 { atomic.AddInt32(&f.pending, -1) } } @@ -95,19 +95,19 @@ func (f *fetchResult) AllDone() bool { // SetReceiptsDone flags the receipts as finished. func (f *fetchResult) SetReceiptsDone() { - if v := atomic.LoadInt32(&f.pending); v == 2 || v == 3 { + if v := atomic.LoadInt32(&f.pending); (v & 2) != 0 { atomic.AddInt32(&f.pending, -2) } } -// CheckDone checks if the given type is done already +// Done 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) + return (v & 1) == 0 case ReceiptType: - return !(v == 2 || v == 3) + return (v & 2) == 0 default: return false } @@ -148,23 +148,19 @@ type queue struct { // newQueue creates a new download queue for scheduling block retrieval. func newQueue(blockCacheLimit int) *queue { lock := new(sync.RWMutex) - return &queue{ - headerPendPool: make(map[string]*fetchRequest), + q := &queue{ headerContCh: make(chan bool), - blockTaskPool: make(map[common.Hash]*types.Header), blockTaskQueue: prque.New(nil), - blockPendPool: make(map[string]*fetchRequest), - receiptTaskPool: make(map[common.Hash]*types.Header), receiptTaskQueue: prque.New(nil), - receiptPendPool: make(map[string]*fetchRequest), - resultCache: newResultStore(blockCacheLimit * 2), active: sync.NewCond(lock), lock: lock, } + q.Reset(blockCacheLimit) + return q } // Reset clears out the queue contents. -func (q *queue) Reset() { +func (q *queue) Reset(blockCacheLimit int) { q.lock.Lock() defer q.lock.Unlock() @@ -182,7 +178,7 @@ func (q *queue) Reset() { q.receiptTaskQueue.Reset() q.receiptPendPool = make(map[string]*fetchRequest) - q.resultCache = newResultStore(blockCacheItems * 2) + q.resultCache = newResultStore(blockCacheLimit * 2) } // Close marks the end of the sync, unblocking Results. @@ -190,8 +186,8 @@ func (q *queue) Reset() { func (q *queue) Close() { q.lock.Lock() q.closed = true + q.active.Signal() q.lock.Unlock() - q.active.Broadcast() } // PendingHeaders retrieves the number of header requests pending for retrieval. @@ -341,20 +337,15 @@ 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) +// Results can be called concurrently with Deliver and Schedule, +// but assumes that there are not two simultaneous callers to Results 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 - } - results := q.resultCache.GetCompleted(maxResultsProcess) - if len(results) == 0 && !block { + if !block && !q.resultCache.HasCompletedItems() { return nil } closed := false - for !closed && len(results) == 0 { + for !closed && !q.resultCache.HasCompletedItems() { // In order to wait on 'active', we need to obtain the lock. // That may take a while, if someone is delivering at the same // time, so after obtaining the lock, we check again if there @@ -363,17 +354,17 @@ func (q *queue) Results(block bool) []*fetchResult { // someone can have closed the queue. In that case, we should // return the available results and stop blocking q.lock.Lock() - closed = q.closed - results = q.resultCache.GetCompleted(maxResultsProcess) - if closed || len(results) > 0 { + if q.resultCache.HasCompletedItems() || q.closed { q.lock.Unlock() break } + // No items available, and not closed q.active.Wait() closed = q.closed q.lock.Unlock() - results = q.resultCache.GetCompleted(maxResultsProcess) } + // Regardless if closed or not, we can still deliver whatever we have + results := q.resultCache.GetCompleted(maxResultsProcess) for _, result := range results { // Recalculate the result item weights to prevent memory exhaustion size := result.Header.Size() @@ -397,7 +388,7 @@ func (q *queue) Results(block bool) []*fetchResult { if time.Now().Second()&0xa == 0 { info := q.Stats() info = append(info, "throttle", throttleThreshold) - log.Info("queue stats", info...) + log.Info("Downloader queue stats", info...) } return results } @@ -866,6 +857,7 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, hashes = append(hashes, header.Hash()) i++ } + q.lock.Lock() var acceptCount = 0 for _, header := range request.Headers[:i] { @@ -889,12 +881,11 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQueue.Push(header, -int64(header.Number.Uint64())) } } - q.lock.Unlock() - // Wake up Results if acceptCount > 0 { - q.active.Broadcast() + q.active.Signal() } + q.lock.Unlock() // If none of the data was good, it's a stale delivery switch { case failure == nil || failure == errInvalidChain: diff --git a/eth/downloader/queue_test.go b/eth/downloader/queue_test.go index f0bdbfc8e0..dc48347d96 100644 --- a/eth/downloader/queue_test.go +++ b/eth/downloader/queue_test.go @@ -242,11 +242,11 @@ func TestEmptyBlocks(t *testing.T) { } -// xTestDelivery does some more extensive testing of events that happen, +// XTestDelivery does some more extensive testing of events that happen, // blocks that become known and peers that make reservations and deliveries. // disabled since it's not really a unit-test, but can be executed to test // some more advanced scenarios -func xTestDelivery(t *testing.T) { +func XTestDelivery(t *testing.T) { // the outside network, holding blocks blo, rec := makeChain(128, 0, genesis, false) world := newNetwork() diff --git a/eth/downloader/resultcache.go b/eth/downloader/resultcache.go index a2cce1924e..be7b8a3a8e 100644 --- a/eth/downloader/resultcache.go +++ b/eth/downloader/resultcache.go @@ -25,7 +25,6 @@ import ( "sync/atomic" "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/log" ) type resultStore struct { @@ -73,31 +72,10 @@ func (r *resultStore) SetThrottleThreshold(threshold uint64) { // prio right now // fetchResult -- the result to store data into // err -- any error that occurred -func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (bool, bool, *fetchResult, error) { - r.lock.RLock() - var index int - item, index, stale, throttled, err := r.getFetchResult(header.Number.Uint64()) - if err != nil || stale || throttled { - r.lock.RUnlock() - // Index is above the current threshold of 'prioritized' blocks, - if throttled { - log.Debug("resultcache throttle", "index", index, "threshold", r.throttleThreshold) - - } - return stale, throttled, item, err - } - if item != nil { - // All good, item already exists (perhaps a receipt fetch following - // a body fetch) - r.lock.RUnlock() - return stale, throttled, item, err - } - 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 +func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale, throttled bool, item *fetchResult, err error) { r.lock.Lock() defer r.lock.Unlock() - // Same checks as above, now with wlock + var index int item, index, stale, throttled, err = r.getFetchResult(header.Number.Uint64()) if err != nil || stale || throttled { return stale, throttled, item, err @@ -132,13 +110,13 @@ func (r *resultStore) getFetchResult(headerNumber uint64) (item *fetchResult, in err = fmt.Errorf("index allocation went beyond available resultStore space "+ "(index [%d] = header [%d] - resultOffset [%d], len(resultStore) = %d", index, headerNumber, r.resultOffset, len(r.items)) - return + return nil, index, stale, throttle, err } if stale { - return + return nil, index, stale, throttle, nil } item = r.items[index] - return + return item, index, stale, throttle, nil } // hasCompletedItems returns true if there are processable items available