diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index bfc7a39a2d..6b496b7eff 100644 --- a/eth/downloader/downloader.go +++ b/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 } } diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index e2eef750c4..c58031955e 100644 --- a/eth/downloader/queue.go +++ b/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()