eth/downloader: detect and skip fetching empty blocks

eth/downloader/downloader.go

@@ -319,7 +319,9 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e
 		}
 	}()
 
-	glog.V(logger.Debug).Infof("Synchronizing with the network using: %s, eth/%d", p.id, p.version)
+	glog.V(logger.Debug).Infof("Synchronising with the network using: %s [eth/%d]", p.id, p.version)
+	defer glog.V(logger.Debug).Infof("Synchronisation terminated")
+
 	switch {
 	case p.version == eth61:
 		// Old eth/61, use forward, concurrent hash and block retrieval algorithm
@@ -887,6 +889,7 @@ func (d *Downloader) findAncestor(p *peer) (uint64, error) {
 // are returned, potentially throttling on the way.
 func (d *Downloader) fetchHeaders(p *peer, td *big.Int, from uint64) error {
 	glog.V(logger.Debug).Infof("%v: downloading headers from #%d", p, from)
+	defer glog.V(logger.Debug).Infof("%v: header download terminated", p)
 
 	// Create a timeout timer, and the associated hash fetcher
 	timeout := time.NewTimer(0) // timer to dump a non-responsive active peer
@@ -1103,6 +1106,7 @@ func (d *Downloader) fetchBodies(from uint64) error {
 				break
 			}
 			// Send a download request to all idle peers, until throttled
+			queuedEmptyBlocks := false
 			for _, peer := range d.peers.IdlePeers() {
 				// Short circuit if throttling activated
 				if d.queue.Throttle() {
@@ -1111,7 +1115,14 @@ func (d *Downloader) fetchBodies(from uint64) error {
 				// Reserve a chunk of hashes for a peer. A nil can mean either that
 				// no more hashes are available, or that the peer is known not to
 				// have them.
-				request := d.queue.Reserve(peer, peer.Capacity())
+				request, process, err := d.queue.Reserve(peer, peer.Capacity())
+				if err != nil {
+					return err
+				}
+				if process {
+					queuedEmptyBlocks = true
+					go d.process()
+				}
 				if request == nil {
 					continue
 				}
@@ -1126,7 +1137,7 @@ func (d *Downloader) fetchBodies(from uint64) error {
 			}
 			// Make sure that we have peers available for fetching. If all peers have been tried
 			// and all failed throw an error
-			if !d.queue.Throttle() && d.queue.InFlight() == 0 {
+			if !queuedEmptyBlocks && !d.queue.Throttle() && d.queue.InFlight() == 0 {
 				return errPeersUnavailable
 			}
 		}

eth/downloader/queue.go

@@ -312,18 +312,19 @@ func (q *queue) Reserve61(p *peer, count int) *fetchRequest {
 }
 
 // Reserve reserves a set of headers for the given peer, skipping any previously
-// failed download.
-func (q *queue) Reserve(p *peer, count int) *fetchRequest {
+// 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) {
 	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() {
-		return nil
+		return nil, false, nil
 	}
 	if _, ok := q.pendPool[p.id]; ok {
-		return nil
+		return nil, false, nil
 	}
 	// Calculate an upper limit on the bodies we might fetch (i.e. throttling)
 	space := len(q.blockCache) - len(q.blockPool)
@@ -334,8 +335,20 @@ func (q *queue) Reserve(p *peer, count int) *fetchRequest {
 	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)
+
+		// 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
+			}
+			delete(q.headerPool, header.Hash())
+			process, space, proc = true, space-1, proc-1
+			continue
+		}
+		// If it's a content block, add to the body fetch request
 		if p.ignored.Has(header.Hash()) {
 			skip = append(skip, header)
 		} else {
@@ -348,7 +361,7 @@ func (q *queue) Reserve(p *peer, count int) *fetchRequest {
 	}
 	// Assemble and return the block download request
 	if len(send) == 0 {
-		return nil
+		return nil, process, nil
 	}
 	request := &fetchRequest{
 		Peer:    p,
@@ -357,7 +370,7 @@ func (q *queue) Reserve(p *peer, count int) *fetchRequest {
 	}
 	q.pendPool[p.id] = request
 
-	return request
+	return request, process, nil
 }
 
 // Cancel aborts a fetch request, returning all pending hashes to the queue.
@@ -424,19 +437,12 @@ func (q *queue) Deliver61(id string, blocks []*types.Block) (err error) {
 			errs = append(errs, fmt.Errorf("non-requested block %x", hash))
 			continue
 		}
-		// 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
-		}
-		// Otherwise merge the block and mark the hash block
-		q.blockCache[index] = &Block{
-			RawBlock:   block,
-			OriginPeer: id,
+		// Queue the block up for processing
+		if err := q.enqueue(id, block); err != nil {
+			return err
 		}
 		delete(request.Hashes, hash)
 		delete(q.hashPool, hash)
-		q.blockPool[hash] = block.NumberU64()
 	}
 	// Return all failed or missing fetches to the queue
 	for hash, index := range request.Hashes {
@@ -483,19 +489,12 @@ func (q *queue) Deliver(id string, transactions [][]*types.Transaction, uncles [
 		}
 		block := types.NewBlockWithHeader(header).WithBody(transactions[i], uncles[i])
 
-		// 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
-		}
-		// Otherwise merge the block and mark the hash done
-		q.blockCache[index] = &Block{
-			RawBlock:   block,
-			OriginPeer: id,
+		// Queue the block up for processing
+		if err := q.enqueue(id, block); err != nil {
+			return err
 		}
 		request.Headers[i] = nil
 		delete(q.headerPool, header.Hash())
-		q.blockPool[header.Hash()] = block.NumberU64()
 	}
 	// Return all failed or missing fetches to the queue
 	for _, header := range request.Headers {
@@ -513,6 +512,23 @@ func (q *queue) Deliver(id string, transactions [][]*types.Transaction, uncles [
 	return nil
 }
 
+// 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
+	}
+	// 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
+}
+
 // Prepare configures the block cache offset to allow accepting inbound blocks.
 func (q *queue) Prepare(offset uint64) {
 	q.lock.Lock()