eth/downloader: add fast and light sync strategies

eth/downloader/queue.go

@@ -29,11 +29,12 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/logger"
 	"github.com/ethereum/go-ethereum/logger/glog"
+	"github.com/rcrowley/go-metrics"
 	"gopkg.in/karalabe/cookiejar.v2/collections/prque"
 )
 
 var (
-	blockCacheLimit = 8 * MaxBlockFetch // Maximum number of blocks to cache before throttling the download
+	blockCacheLimit = 1024 // Maximum number of blocks to cache before throttling the download
 )
 
 var (
@@ -41,29 +42,47 @@ var (
 	errStaleDelivery    = errors.New("stale delivery")
 )
 
-// fetchRequest is a currently running block retrieval operation.
+// fetchRequest is a currently running data retrieval operation.
 type fetchRequest struct {
 	Peer    *peer               // Peer to which the request was sent
-	Hashes  map[common.Hash]int // [eth/61] Requested hashes with their insertion index (priority)
+	Hashes  map[common.Hash]int // [eth/61] Requested block with their insertion index (priority)
 	Headers []*types.Header     // [eth/62] Requested headers, sorted by request order
 	Time    time.Time           // Time when the request was made
 }
 
+// fetchResult is the assembly collecting partial results from potentially more
+// than one fetcher routines, until all outstanding retrievals complete and the
+// result as a whole can be processed.
+type fetchResult struct {
+	Pending int // Number of data fetches still pending
+
+	Header       *types.Header
+	Uncles       []*types.Header
+	Transactions types.Transactions
+	Receipts     types.Receipts
+}
+
 // queue represents hashes that are either need fetching or are being fetched
 type queue struct {
 	hashPool    map[common.Hash]int // [eth/61] Pending hashes, mapping to their insertion index (priority)
 	hashQueue   *prque.Prque        // [eth/61] Priority queue of the block hashes to fetch
 	hashCounter int                 // [eth/61] Counter indexing the added hashes to ensure retrieval order
 
-	headerPool  map[common.Hash]*types.Header // [eth/62] Pending headers, mapping from their hashes
-	headerQueue *prque.Prque                  // [eth/62] Priority queue of the headers to fetch the bodies for
-	headerHead  common.Hash                   // [eth/62] Hash of the last queued header to verify order
+	headerHead common.Hash // [eth/62] Hash of the last queued header to verify order
 
-	pendPool map[string]*fetchRequest // Currently pending block retrieval operations
+	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
 
-	blockPool   map[common.Hash]uint64 // Hash-set of the downloaded data blocks, mapping to cache indexes
-	blockCache  []*Block               // Downloaded but not yet delivered blocks
-	blockOffset uint64                 // Offset of the first cached block in the block-chain
+	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  []*fetchResult // Downloaded but not yet delivered fetch results
+	resultOffset uint64         // Offset of the first cached fetch result in the block-chain
+	resultParts  int            // Number of fetch components required to complete an item
 
 	lock sync.RWMutex
 }
@@ -71,13 +90,17 @@ type queue struct {
 // newQueue creates a new download queue for scheduling block retrieval.
 func newQueue() *queue {
 	return &queue{
-		hashPool:    make(map[common.Hash]int),
-		hashQueue:   prque.New(),
-		headerPool:  make(map[common.Hash]*types.Header),
-		headerQueue: prque.New(),
-		pendPool:    make(map[string]*fetchRequest),
-		blockPool:   make(map[common.Hash]uint64),
-		blockCache:  make([]*Block, blockCacheLimit),
+		hashPool:         make(map[common.Hash]int),
+		hashQueue:        prque.New(),
+		blockTaskPool:    make(map[common.Hash]*types.Header),
+		blockTaskQueue:   prque.New(),
+		blockPendPool:    make(map[string]*fetchRequest),
+		blockDonePool:    make(map[common.Hash]struct{}),
+		receiptTaskPool:  make(map[common.Hash]*types.Header),
+		receiptTaskQueue: prque.New(),
+		receiptPendPool:  make(map[string]*fetchRequest),
+		receiptDonePool:  make(map[common.Hash]struct{}),
+		resultCache:      make([]*fetchResult, blockCacheLimit),
 	}
 }
 
@@ -90,32 +113,37 @@ func (q *queue) Reset() {
 	q.hashQueue.Reset()
 	q.hashCounter = 0
 
-	q.headerPool = make(map[common.Hash]*types.Header)
-	q.headerQueue.Reset()
 	q.headerHead = common.Hash{}
 
-	q.pendPool = make(map[string]*fetchRequest)
+	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.blockPool = make(map[common.Hash]uint64)
-	q.blockOffset = 0
-	q.blockCache = make([]*Block, blockCacheLimit)
+	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 = make([]*fetchResult, blockCacheLimit)
+	q.resultOffset = 0
+	q.resultParts = 0
 }
 
-// Size retrieves the number of blocks in the queue, returning separately for
-// pending and already downloaded.
-func (q *queue) Size() (int, int) {
+// PendingBlocks retrieves the number of block (body) requests pending for retrieval.
+func (q *queue) PendingBlocks() int {
 	q.lock.RLock()
 	defer q.lock.RUnlock()
 
-	return len(q.hashPool) + len(q.headerPool), len(q.blockPool)
+	return q.hashQueue.Size() + q.blockTaskQueue.Size()
 }
 
-// Pending retrieves the number of blocks pending for retrieval.
-func (q *queue) Pending() int {
+// PendingReceipts retrieves the number of block receipts pending for retrieval.
+func (q *queue) PendingReceipts() int {
 	q.lock.RLock()
 	defer q.lock.RUnlock()
 
-	return q.hashQueue.Size() + q.headerQueue.Size()
+	return q.receiptTaskQueue.Size()
 }
 
 // InFlight retrieves the number of fetch requests currently in flight.
@@ -123,44 +151,55 @@ func (q *queue) InFlight() int {
 	q.lock.RLock()
 	defer q.lock.RUnlock()
 
-	return len(q.pendPool)
+	return len(q.blockPendPool) + len(q.receiptPendPool)
 }
 
-// Throttle checks if the download should be throttled (active block fetches
-// exceed block cache).
-func (q *queue) Throttle() bool {
+// Idle returns if the queue is fully idle or has some data still inside. This
+// method is used by the tester to detect termination events.
+func (q *queue) Idle() bool {
 	q.lock.RLock()
 	defer q.lock.RUnlock()
 
-	// Calculate the currently in-flight block requests
+	queued := q.hashQueue.Size() + 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
+}
+
+// ThrottleBlocks checks if the download should be throttled (active block (body)
+// fetches exceed block cache).
+func (q *queue) ThrottleBlocks() bool {
+	q.lock.RLock()
+	defer q.lock.RUnlock()
+
+	// Calculate the currently in-flight block (body) requests
 	pending := 0
-	for _, request := range q.pendPool {
+	for _, request := range q.blockPendPool {
 		pending += len(request.Hashes) + len(request.Headers)
 	}
-	// Throttle if more blocks are in-flight than free space in the cache
-	return pending >= len(q.blockCache)-len(q.blockPool)
+	// Throttle if more blocks (bodies) are in-flight than free space in the cache
+	return pending >= len(q.resultCache)-len(q.blockDonePool)
 }
 
-// Has checks if a hash is within the download queue or not.
-func (q *queue) Has(hash common.Hash) bool {
+// ThrottleReceipts checks if the download should be throttled (active receipt
+// fetches exceed block cache).
+func (q *queue) ThrottleReceipts() bool {
 	q.lock.RLock()
 	defer q.lock.RUnlock()
 
-	if _, ok := q.hashPool[hash]; ok {
-		return true
+	// Calculate the currently in-flight receipt requests
+	pending := 0
+	for _, request := range q.receiptPendPool {
+		pending += len(request.Headers)
 	}
-	if _, ok := q.headerPool[hash]; ok {
-		return true
-	}
-	if _, ok := q.blockPool[hash]; ok {
-		return true
-	}
-	return false
+	// Throttle if more receipts are in-flight than free space in the cache
+	return pending >= len(q.resultCache)-len(q.receiptDonePool)
 }
 
-// Insert61 adds a set of hashes for the download queue for scheduling, returning
+// Schedule61 adds a set of hashes for the download queue for scheduling, returning
 // the new hashes encountered.
-func (q *queue) Insert61(hashes []common.Hash, fifo bool) []common.Hash {
+func (q *queue) Schedule61(hashes []common.Hash, fifo bool) []common.Hash {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
@@ -186,22 +225,17 @@ func (q *queue) Insert61(hashes []common.Hash, fifo bool) []common.Hash {
 	return inserts
 }
 
-// Insert adds a set of headers for the download queue for scheduling, returning
+// Schedule adds a set of headers for the download queue for scheduling, returning
 // the new headers encountered.
-func (q *queue) Insert(headers []*types.Header, from uint64) []*types.Header {
+func (q *queue) Schedule(headers []*types.Header, from uint64, receipts bool) []*types.Header {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
 	// Insert all the headers prioritized by the contained block number
 	inserts := make([]*types.Header, 0, len(headers))
 	for _, header := range headers {
-		// Make sure no duplicate requests are executed
-		hash := header.Hash()
-		if _, ok := q.headerPool[hash]; ok {
-			glog.V(logger.Warn).Infof("Header #%d [%x] already scheduled", header.Number.Uint64(), hash[:4])
-			continue
-		}
 		// Make sure chain order is honored and preserved throughout
+		hash := header.Hash()
 		if header.Number == nil || header.Number.Uint64() != from {
 			glog.V(logger.Warn).Infof("Header #%v [%x] broke chain ordering, expected %d", header.Number, hash[:4], from)
 			break
@@ -210,69 +244,72 @@ func (q *queue) Insert(headers []*types.Header, from uint64) []*types.Header {
 			glog.V(logger.Warn).Infof("Header #%v [%x] broke chain ancestry", header.Number, hash[:4])
 			break
 		}
-		// Queue the header for body retrieval
+		// Make sure no duplicate requests are executed
+		if _, ok := q.blockTaskPool[hash]; ok {
+			glog.V(logger.Warn).Infof("Header #%d [%x] already scheduled for block fetch", header.Number.Uint64(), hash[:4])
+			continue
+		}
+		if _, ok := q.receiptTaskPool[hash]; ok {
+			glog.V(logger.Warn).Infof("Header #%d [%x] already scheduled for receipt fetch", header.Number.Uint64(), hash[:4])
+			continue
+		}
+		// Queue the header for content retrieval
+		q.blockTaskPool[hash] = header
+		q.blockTaskQueue.Push(header, -float32(header.Number.Uint64()))
+		if receipts {
+			q.receiptTaskPool[hash] = header
+			q.receiptTaskQueue.Push(header, -float32(header.Number.Uint64()))
+		}
 		inserts = append(inserts, header)
-		q.headerPool[hash] = header
-		q.headerQueue.Push(header, -float32(header.Number.Uint64()))
 		q.headerHead = hash
 		from++
 	}
 	return inserts
 }
 
-// GetHeadBlock retrieves the first block from the cache, or nil if it hasn't
+// GetHeadResult retrieves the first fetch result from the cache, or nil if it hasn't
 // been downloaded yet (or simply non existent).
-func (q *queue) GetHeadBlock() *Block {
+func (q *queue) GetHeadResult() *fetchResult {
 	q.lock.RLock()
 	defer q.lock.RUnlock()
 
-	if len(q.blockCache) == 0 {
+	if len(q.resultCache) == 0 || q.resultCache[0] == nil {
 		return nil
 	}
-	return q.blockCache[0]
-}
-
-// GetBlock retrieves a downloaded block, or nil if non-existent.
-func (q *queue) GetBlock(hash common.Hash) *Block {
-	q.lock.RLock()
-	defer q.lock.RUnlock()
-
-	// Short circuit if the block hasn't been downloaded yet
-	index, ok := q.blockPool[hash]
-	if !ok {
+	if q.resultCache[0].Pending > 0 {
 		return nil
 	}
-	// Return the block if it's still available in the cache
-	if q.blockOffset <= index && index < q.blockOffset+uint64(len(q.blockCache)) {
-		return q.blockCache[index-q.blockOffset]
-	}
-	return nil
+	return q.resultCache[0]
 }
 
-// TakeBlocks retrieves and permanently removes a batch of blocks from the cache.
-func (q *queue) TakeBlocks() []*Block {
+// TakeResults retrieves and permanently removes a batch of fetch results from
+// the cache.
+func (q *queue) TakeResults() []*fetchResult {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
-	// Accumulate all available blocks
-	blocks := []*Block{}
-	for _, block := range q.blockCache {
-		if block == nil {
+	// Accumulate all available results
+	results := []*fetchResult{}
+	for _, result := range q.resultCache {
+		if result == nil || result.Pending > 0 {
 			break
 		}
-		blocks = append(blocks, block)
-		delete(q.blockPool, block.RawBlock.Hash())
-	}
-	// Delete the blocks from the slice and let them be garbage collected
-	// without this slice trick the blocks would stay in memory until nil
-	// would be assigned to q.blocks
-	copy(q.blockCache, q.blockCache[len(blocks):])
-	for k, n := len(q.blockCache)-len(blocks), len(q.blockCache); k < n; k++ {
-		q.blockCache[k] = nil
-	}
-	q.blockOffset += uint64(len(blocks))
+		results = append(results, result)
 
-	return blocks
+		hash := result.Header.Hash()
+		delete(q.blockDonePool, hash)
+		delete(q.receiptDonePool, hash)
+	}
+	// Delete the results from the slice and let them be garbage collected
+	// without this slice trick the results would stay in memory until nil
+	// would be assigned to them.
+	copy(q.resultCache, q.resultCache[len(results):])
+	for k, n := len(q.resultCache)-len(results), len(q.resultCache); k < n; k++ {
+		q.resultCache[k] = nil
+	}
+	q.resultOffset += uint64(len(results))
+
+	return results
 }
 
 // Reserve61 reserves a set of hashes for the given peer, skipping any previously
@@ -286,12 +323,12 @@ func (q *queue) Reserve61(p *peer, count int) *fetchRequest {
 	if q.hashQueue.Empty() {
 		return nil
 	}
-	if _, ok := q.pendPool[p.id]; ok {
+	if _, ok := q.blockPendPool[p.id]; ok {
 		return nil
 	}
 	// Calculate an upper limit on the hashes we might fetch (i.e. throttling)
-	space := len(q.blockCache) - len(q.blockPool)
-	for _, request := range q.pendPool {
+	space := len(q.resultCache) - len(q.blockDonePool)
+	for _, request := range q.blockPendPool {
 		space -= len(request.Hashes)
 	}
 	// Retrieve a batch of hashes, skipping previously failed ones
@@ -319,49 +356,82 @@ func (q *queue) Reserve61(p *peer, count int) *fetchRequest {
 		Hashes: send,
 		Time:   time.Now(),
 	}
-	q.pendPool[p.id] = request
+	q.blockPendPool[p.id] = request
 
 	return request
 }
 
-// Reserve reserves a set of headers for the given peer, skipping any previously
-// failed download. Beside the next batch of needed fetches, it also returns a
-// flag whether empty blocks were queued requiring processing.
-func (q *queue) Reserve(p *peer, count int) (*fetchRequest, bool, error) {
+// ReserveBlocks 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) ReserveBlocks(p *peer, count int) (*fetchRequest, bool, error) {
+	noop := func(header *types.Header) bool {
+		return header.TxHash == types.EmptyRootHash && header.UncleHash == types.EmptyUncleHash
+	}
+	return q.reserveFetch(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, noop)
+}
+
+// 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 *peer, count int) (*fetchRequest, bool, error) {
+	noop := func(header *types.Header) bool {
+		return header.ReceiptHash == types.EmptyRootHash
+	}
+	return q.reserveFetch(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, noop)
+}
+
+// reserveFetch reserves a set of data download operations for a given peer,
+// skipping any previously failed ones. This method is a generic version used
+// by the individual special reservation functions.
+func (q *queue) reserveFetch(p *peer, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
+	pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, noop func(*types.Header) bool) (*fetchRequest, bool, error) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
 	// Short circuit if the pool has been depleted, or if the peer's already
 	// downloading something (sanity check not to corrupt state)
-	if q.headerQueue.Empty() {
+	if taskQueue.Empty() {
 		return nil, false, nil
 	}
-	if _, ok := q.pendPool[p.id]; ok {
+	if _, ok := pendPool[p.id]; ok {
 		return nil, false, nil
 	}
-	// Calculate an upper limit on the bodies we might fetch (i.e. throttling)
-	space := len(q.blockCache) - len(q.blockPool)
-	for _, request := range q.pendPool {
+	// Calculate an upper limit on the items we might fetch (i.e. throttling)
+	space := len(q.resultCache) - len(donePool)
+	for _, request := range pendPool {
 		space -= len(request.Headers)
 	}
-	// Retrieve a batch of headers, skipping previously failed ones
+	// Retrieve a batch of tasks, skipping previously failed ones
 	send := make([]*types.Header, 0, count)
 	skip := make([]*types.Header, 0)
 
-	process := false
-	for proc := 0; proc < space && len(send) < count && !q.headerQueue.Empty(); proc++ {
-		header := q.headerQueue.PopItem().(*types.Header)
+	progress := false
+	for proc := 0; proc < space && len(send) < count && !taskQueue.Empty(); proc++ {
+		header := taskQueue.PopItem().(*types.Header)
 
-		// If the header defines an empty block, deliver straight
-		if header.TxHash == types.DeriveSha(types.Transactions{}) && header.UncleHash == types.CalcUncleHash([]*types.Header{}) {
-			if err := q.enqueue("", types.NewBlockWithHeader(header)); err != nil {
-				return nil, false, errInvalidChain
+		// If we're the first to request this task, initialize the result container
+		index := int(header.Number.Int64() - int64(q.resultOffset))
+		if index >= len(q.resultCache) || index < 0 {
+			return nil, false, errInvalidChain
+		}
+		if q.resultCache[index] == nil {
+			q.resultCache[index] = &fetchResult{
+				Pending: q.resultParts,
+				Header:  header,
 			}
-			delete(q.headerPool, header.Hash())
-			process, space, proc = true, space-1, proc-1
+		}
+		// If this fetch task is a noop, skip this fetch operation
+		if noop(header) {
+			donePool[header.Hash()] = struct{}{}
+			delete(taskPool, header.Hash())
+
+			space, proc = space-1, proc-1
+			q.resultCache[index].Pending--
+			progress = true
 			continue
 		}
-		// If it's a content block, add to the body fetch request
+		// Otherwise if not a known unknown block, add to the retrieve list
 		if p.ignored.Has(header.Hash()) {
 			skip = append(skip, header)
 		} else {
@@ -370,24 +440,41 @@ func (q *queue) Reserve(p *peer, count int) (*fetchRequest, bool, error) {
 	}
 	// Merge all the skipped headers back
 	for _, header := range skip {
-		q.headerQueue.Push(header, -float32(header.Number.Uint64()))
+		taskQueue.Push(header, -float32(header.Number.Uint64()))
 	}
 	// Assemble and return the block download request
 	if len(send) == 0 {
-		return nil, process, nil
+		return nil, progress, nil
 	}
 	request := &fetchRequest{
 		Peer:    p,
 		Headers: send,
 		Time:    time.Now(),
 	}
-	q.pendPool[p.id] = request
+	pendPool[p.id] = request
 
-	return request, process, nil
+	return request, progress, nil
 }
 
-// Cancel aborts a fetch request, returning all pending hashes to the queue.
-func (q *queue) Cancel(request *fetchRequest) {
+// Cancel61 aborts a fetch request, returning all pending hashes to the queue.
+func (q *queue) Cancel61(request *fetchRequest) {
+	q.cancel(request, nil, q.blockPendPool)
+}
+
+// CancelBlocks aborts a body fetch request, returning all pending hashes to the
+// task queue.
+func (q *queue) CancelBlocks(request *fetchRequest) {
+	q.cancel(request, q.blockTaskQueue, q.blockPendPool)
+}
+
+// CancelReceipts aborts a body fetch request, returning all pending hashes to
+// the task queue.
+func (q *queue) CancelReceipts(request *fetchRequest) {
+	q.cancel(request, q.receiptTaskQueue, q.receiptPendPool)
+}
+
+// Cancel aborts a fetch request, returning all pending hashes to the task queue.
+func (q *queue) cancel(request *fetchRequest, taskQueue *prque.Prque, pendPool map[string]*fetchRequest) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
@@ -395,20 +482,62 @@ func (q *queue) Cancel(request *fetchRequest) {
 		q.hashQueue.Push(hash, float32(index))
 	}
 	for _, header := range request.Headers {
-		q.headerQueue.Push(header, -float32(header.Number.Uint64()))
+		taskQueue.Push(header, -float32(header.Number.Uint64()))
 	}
-	delete(q.pendPool, request.Peer.id)
+	delete(pendPool, request.Peer.id)
 }
 
-// Expire checks for in flight requests that exceeded a timeout allowance,
+// Revoke cancels all pending requests belonging to a given peer. This method is
+// meant to be called during a peer drop to quickly reassign owned data fetches
+// to remaining nodes.
+func (q *queue) Revoke(peerId string) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	if request, ok := q.blockPendPool[peerId]; ok {
+		for hash, index := range request.Hashes {
+			q.hashQueue.Push(hash, float32(index))
+		}
+		for _, header := range request.Headers {
+			q.blockTaskQueue.Push(header, -float32(header.Number.Uint64()))
+		}
+		delete(q.blockPendPool, peerId)
+	}
+	if request, ok := q.receiptPendPool[peerId]; ok {
+		for _, header := range request.Headers {
+			q.receiptTaskQueue.Push(header, -float32(header.Number.Uint64()))
+		}
+		delete(q.receiptPendPool, peerId)
+	}
+}
+
+// Expire61 checks for in flight requests that exceeded a timeout allowance,
 // canceling them and returning the responsible peers for penalization.
-func (q *queue) Expire(timeout time.Duration) []string {
+func (q *queue) Expire61(timeout time.Duration) []string {
+	return q.expire(timeout, q.blockPendPool, nil)
+}
+
+// ExpireBlocks checks for in flight block body requests that exceeded a timeout
+// allowance, canceling them and returning the responsible peers for penalization.
+func (q *queue) ExpireBlocks(timeout time.Duration) []string {
+	return q.expire(timeout, q.blockPendPool, q.blockTaskQueue)
+}
+
+// ExpireReceipts checks for in flight receipt requests that exceeded a timeout
+// allowance, canceling them and returning the responsible peers for penalization.
+func (q *queue) ExpireReceipts(timeout time.Duration) []string {
+	return q.expire(timeout, q.receiptPendPool, q.receiptTaskQueue)
+}
+
+// expire is the generic check that move expired tasks from a pending pool back
+// into a task pool, returning all entities caught with expired tasks.
+func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest, taskQueue *prque.Prque) []string {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
 	// Iterate over the expired requests and return each to the queue
 	peers := []string{}
-	for id, request := range q.pendPool {
+	for id, request := range pendPool {
 		if time.Since(request.Time) > timeout {
 			// Update the metrics with the timeout
 			if len(request.Hashes) > 0 {
@@ -421,14 +550,14 @@ func (q *queue) Expire(timeout time.Duration) []string {
 				q.hashQueue.Push(hash, float32(index))
 			}
 			for _, header := range request.Headers {
-				q.headerQueue.Push(header, -float32(header.Number.Uint64()))
+				taskQueue.Push(header, -float32(header.Number.Uint64()))
 			}
 			peers = append(peers, id)
 		}
 	}
 	// Remove the expired requests from the pending pool
 	for _, id := range peers {
-		delete(q.pendPool, id)
+		delete(pendPool, id)
 	}
 	return peers
 }
@@ -439,12 +568,12 @@ func (q *queue) Deliver61(id string, blocks []*types.Block) (err error) {
 	defer q.lock.Unlock()
 
 	// Short circuit if the blocks were never requested
-	request := q.pendPool[id]
+	request := q.blockPendPool[id]
 	if request == nil {
 		return errNoFetchesPending
 	}
 	blockReqTimer.UpdateSince(request.Time)
-	delete(q.pendPool, id)
+	delete(q.blockPendPool, id)
 
 	// If no blocks were retrieved, mark them as unavailable for the origin peer
 	if len(blocks) == 0 {
@@ -461,10 +590,19 @@ func (q *queue) Deliver61(id string, blocks []*types.Block) (err error) {
 			errs = append(errs, fmt.Errorf("non-requested block %x", hash))
 			continue
 		}
-		// Queue the block up for processing
-		if err := q.enqueue(id, block); err != nil {
-			return err
+		// Reconstruct the next result if contents match up
+		index := int(block.Number().Int64() - int64(q.resultOffset))
+		if index >= len(q.resultCache) || index < 0 {
+			errs = []error{errInvalidChain}
+			break
 		}
+		q.resultCache[index] = &fetchResult{
+			Header:       block.Header(),
+			Transactions: block.Transactions(),
+			Uncles:       block.Uncles(),
+		}
+		q.blockDonePool[block.Hash()] = struct{}{}
+
 		delete(request.Hashes, hash)
 		delete(q.hashPool, hash)
 	}
@@ -473,72 +611,12 @@ func (q *queue) Deliver61(id string, blocks []*types.Block) (err error) {
 		q.hashQueue.Push(hash, float32(index))
 	}
 	// If none of the blocks were good, it's a stale delivery
-	if len(errs) != 0 {
-		if len(errs) == len(blocks) {
-			return errStaleDelivery
-		}
-		return fmt.Errorf("multiple failures: %v", errs)
-	}
-	return nil
-}
-
-// Deliver injects a block body retrieval response into the download queue.
-func (q *queue) Deliver(id string, txLists [][]*types.Transaction, uncleLists [][]*types.Header) error {
-	q.lock.Lock()
-	defer q.lock.Unlock()
-
-	// Short circuit if the block bodies were never requested
-	request := q.pendPool[id]
-	if request == nil {
-		return errNoFetchesPending
-	}
-	bodyReqTimer.UpdateSince(request.Time)
-	delete(q.pendPool, id)
-
-	// If no block bodies were retrieved, mark them as unavailable for the origin peer
-	if len(txLists) == 0 || len(uncleLists) == 0 {
-		for hash, _ := range request.Headers {
-			request.Peer.ignored.Add(hash)
-		}
-	}
-	// Assemble each of the block bodies with their headers and queue for processing
-	errs := make([]error, 0)
-	for i, header := range request.Headers {
-		// Short circuit block assembly if no more bodies are found
-		if i >= len(txLists) || i >= len(uncleLists) {
-			break
-		}
-		// Reconstruct the next block if contents match up
-		if types.DeriveSha(types.Transactions(txLists[i])) != header.TxHash || types.CalcUncleHash(uncleLists[i]) != header.UncleHash {
-			errs = []error{errInvalidBody}
-			break
-		}
-		block := types.NewBlockWithHeader(header).WithBody(txLists[i], uncleLists[i])
-
-		// Queue the block up for processing
-		if err := q.enqueue(id, block); err != nil {
-			errs = []error{err}
-			break
-		}
-		request.Headers[i] = nil
-		delete(q.headerPool, header.Hash())
-	}
-	// Return all failed or missing fetches to the queue
-	for _, header := range request.Headers {
-		if header != nil {
-			q.headerQueue.Push(header, -float32(header.Number.Uint64()))
-		}
-	}
-	// If none of the blocks were good, it's a stale delivery
 	switch {
 	case len(errs) == 0:
 		return nil
 
-	case len(errs) == 1 && errs[0] == errInvalidBody:
-		return errInvalidBody
-
-	case len(errs) == 1 && errs[0] == errInvalidChain:
-		return errInvalidChain
+	case len(errs) == 1 && (errs[0] == errInvalidChain || errs[0] == errInvalidBlock):
+		return errs[0]
 
 	case len(errs) == len(request.Headers):
 		return errStaleDelivery
@@ -548,29 +626,105 @@ func (q *queue) Deliver(id string, txLists [][]*types.Transaction, uncleLists []
 	}
 }
 
-// enqueue inserts a new block into the final delivery queue, waiting for pickup
-// by the processor.
-func (q *queue) enqueue(origin string, block *types.Block) error {
-	// If a requested block falls out of the range, the hash chain is invalid
-	index := int(int64(block.NumberU64()) - int64(q.blockOffset))
-	if index >= len(q.blockCache) || index < 0 {
-		return errInvalidChain
+// DeliverBlocks injects a block (body) retrieval response into the results queue.
+func (q *queue) DeliverBlocks(id string, txLists [][]*types.Transaction, uncleLists [][]*types.Header) error {
+	reconstruct := func(header *types.Header, index int, result *fetchResult) error {
+		if types.DeriveSha(types.Transactions(txLists[index])) != header.TxHash || types.CalcUncleHash(uncleLists[index]) != header.UncleHash {
+			return errInvalidBody
+		}
+		result.Transactions = txLists[index]
+		result.Uncles = uncleLists[index]
+		return nil
 	}
-	// Otherwise merge the block and mark the hash done
-	q.blockCache[index] = &Block{
-		RawBlock:   block,
-		OriginPeer: origin,
-	}
-	q.blockPool[block.Header().Hash()] = block.NumberU64()
-	return nil
+	return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, bodyReqTimer, len(txLists), reconstruct)
 }
 
-// Prepare configures the block cache offset to allow accepting inbound blocks.
-func (q *queue) Prepare(offset uint64) {
+// DeliverReceipts injects a receipt retrieval response into the results queue.
+func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) error {
+	reconstruct := func(header *types.Header, index int, result *fetchResult) error {
+		if types.DeriveSha(types.Receipts(receiptList[index])) != header.ReceiptHash {
+			return errInvalidReceipt
+		}
+		result.Receipts = receiptList[index]
+		return nil
+	}
+	return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, receiptReqTimer, len(receiptList), reconstruct)
+}
+
+// deliver injects a data retrieval response into the results queue.
+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, reconstruct func(header *types.Header, index int, result *fetchResult) error) error {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
-	if q.blockOffset < offset {
-		q.blockOffset = offset
+	// Short circuit if the data was never requested
+	request := pendPool[id]
+	if request == nil {
+		return errNoFetchesPending
+	}
+	reqTimer.UpdateSince(request.Time)
+	delete(pendPool, id)
+
+	// If no data items were retrieved, mark them as unavailable for the origin peer
+	if results == 0 {
+		for hash, _ := range request.Headers {
+			request.Peer.ignored.Add(hash)
+		}
+	}
+	// Assemble each of the results with their headers and retrieved data parts
+	errs := make([]error, 0)
+	for i, header := range request.Headers {
+		// Short circuit assembly if no more fetch results are found
+		if i >= results {
+			break
+		}
+		// Reconstruct the next result if contents match up
+		index := int(header.Number.Int64() - int64(q.resultOffset))
+		if index >= len(q.resultCache) || index < 0 || q.resultCache[index] == nil {
+			errs = []error{errInvalidChain}
+			break
+		}
+		if err := reconstruct(header, i, q.resultCache[index]); err != nil {
+			errs = []error{err}
+			break
+		}
+		donePool[header.Hash()] = struct{}{}
+		q.resultCache[index].Pending--
+
+		// Clean up a successful fetch
+		request.Headers[i] = nil
+		delete(taskPool, header.Hash())
+	}
+	// Return all failed or missing fetches to the queue
+	for _, header := range request.Headers {
+		if header != nil {
+			taskQueue.Push(header, -float32(header.Number.Uint64()))
+		}
+	}
+	// If none of the blocks were good, it's a stale delivery
+	switch {
+	case len(errs) == 0:
+		return nil
+
+	case len(errs) == 1 && (errs[0] == errInvalidChain || errs[0] == errInvalidBody || errs[0] == errInvalidReceipt):
+		return errs[0]
+
+	case len(errs) == len(request.Headers):
+		return errStaleDelivery
+
+	default:
+		return fmt.Errorf("multiple failures: %v", errs)
 	}
 }
+
+// Prepare configures the result cache to allow accepting and caching inbound
+// fetch results.
+func (q *queue) Prepare(offset uint64, parts int) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	if q.resultOffset < offset {
+		q.resultOffset = offset
+	}
+	q.resultParts = parts
+}