eth, eth/downloader: move block processing into the downlaoder

eth/downloader/downloader.go

@@ -3,6 +3,7 @@ package downloader
 import (
 	"bytes"
 	"errors"
+	"math"
 	"math/rand"
 	"sync"
 	"sync/atomic"
@@ -28,6 +29,7 @@ var (
 	crossCheckCycle = time.Second      // Period after which to check for expired cross checks
 
 	maxBannedHashes = 4096 // Number of bannable hashes before phasing old ones out
+	maxBlockProcess = 256  // Number of blocks to import at once into the chain
 )
 
 var (
@@ -44,8 +46,9 @@ var (
 	errAlreadyInPool     = errors.New("hash already in pool")
 	errInvalidChain      = errors.New("retrieved hash chain is invalid")
 	errCrossCheckFailed  = errors.New("block cross-check failed")
-	errCancelHashFetch  = errors.New("hash fetching cancelled (requested)")
-	errCancelBlockFetch = errors.New("block downloading cancelled (requested)")
+	errCancelHashFetch   = errors.New("hash fetching canceled (requested)")
+	errCancelBlockFetch  = errors.New("block downloading canceled (requested)")
+	errCancelChainImport = errors.New("chain importing canceled (requested)")
 	errNoSyncActive      = errors.New("no sync active")
 )
 
@@ -55,6 +58,9 @@ type hashCheckFn func(common.Hash) bool
 // blockRetrievalFn is a callback type for retrieving a block from the local chain.
 type blockRetrievalFn func(common.Hash) *types.Block
 
+// chainInsertFn is a callback type to insert a batch of blocks into the local chain.
+type chainInsertFn func(types.Blocks) (int, error)
+
 // peerDropFn is a callback type for dropping a peer detected as malicious.
 type peerDropFn func(id string)
 
@@ -90,11 +96,13 @@ type Downloader struct {
 	// Callbacks
 	hasBlock    hashCheckFn      // Checks if a block is present in the chain
 	getBlock    blockRetrievalFn // Retrieves a block from the chain
+	insertChain chainInsertFn    // Injects a batch of blocks into the chain
 	dropPeer    peerDropFn       // Retrieved the TD of our own chain
 
 	// Status
 	synchroniseMock func(id string, hash common.Hash) error // Replacement for synchronise during testing
 	synchronising   int32
+	processing      int32
 	notified        int32
 
 	// Channels
@@ -113,7 +121,7 @@ type Block struct {
 }
 
 // New creates a new downloader to fetch hashes and blocks from remote peers.
-func New(mux *event.TypeMux, hasBlock hashCheckFn, getBlock blockRetrievalFn, dropPeer peerDropFn) *Downloader {
+func New(mux *event.TypeMux, hasBlock hashCheckFn, getBlock blockRetrievalFn, insertChain chainInsertFn, dropPeer peerDropFn) *Downloader {
 	// Create the base downloader
 	downloader := &Downloader{
 		mux:         mux,
@@ -121,6 +129,7 @@ func New(mux *event.TypeMux, hasBlock hashCheckFn, getBlock blockRetrievalFn, dr
 		peers:       newPeerSet(),
 		hasBlock:    hasBlock,
 		getBlock:    getBlock,
+		insertChain: insertChain,
 		dropPeer:    dropPeer,
 		newPeerCh:   make(chan *peer, 1),
 		hashCh:      make(chan hashPack, 1),
@@ -157,7 +166,7 @@ func (d *Downloader) Stats() (pending int, cached int, importing int, estimate t
 	return
 }
 
-// Synchronising returns the state of the downloader
+// Synchronising returns whether the downloader is currently retrieving blocks.
 func (d *Downloader) Synchronising() bool {
 	return atomic.LoadInt32(&d.synchronising) > 0
 }
@@ -260,19 +269,6 @@ func (d *Downloader) synchronise(id string, hash common.Hash) error {
 	return d.syncWithPeer(p, hash)
 }
 
-// TakeBlocks takes blocks from the queue and yields them to the caller.
-func (d *Downloader) TakeBlocks() []*Block {
-	blocks := d.queue.TakeBlocks()
-	if len(blocks) > 0 {
-		d.importLock.Lock()
-		d.importStart = time.Now()
-		d.importQueue = blocks
-		d.importDone = 0
-		d.importLock.Unlock()
-	}
-	return blocks
-}
-
 // Has checks if the downloader knows about a particular hash, meaning that its
 // either already downloaded of pending retrieval.
 func (d *Downloader) Has(hash common.Hash) bool {
@@ -307,20 +303,17 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash) (err error) {
 
 // Cancel cancels all of the operations and resets the queue. It returns true
 // if the cancel operation was completed.
-func (d *Downloader) Cancel() bool {
-	// If we're not syncing just return.
-	hs, bs := d.queue.Size()
-	if atomic.LoadInt32(&d.synchronising) == 0 && hs == 0 && bs == 0 {
-		return false
-	}
+func (d *Downloader) Cancel() {
 	// Close the current cancel channel
 	d.cancelLock.Lock()
+	if d.cancelCh != nil {
 		select {
 		case <-d.cancelCh:
 			// Channel was already closed
 		default:
 			close(d.cancelCh)
 		}
+	}
 	d.cancelLock.Unlock()
 
 	// Reset the queue and import statistics
@@ -330,11 +323,11 @@ func (d *Downloader) Cancel() bool {
 	d.importQueue = nil
 	d.importDone = 0
 	d.importLock.Unlock()
-
-	return true
 }
 
-// XXX Make synchronous
+// fetchHahes starts retrieving hashes backwards from a specific peer and hash,
+// up until it finds a common ancestor. If the source peer times out, alternative
+// ones are tried for continuation.
 func (d *Downloader) fetchHashes(p *peer, h common.Hash) error {
 	var (
 		start  = time.Now()
@@ -530,10 +523,13 @@ out:
 						glog.V(logger.Detail).Infof("%s: no blocks delivered", peer)
 						break
 					}
-					// All was successful, promote the peer
+					// All was successful, promote the peer and potentially start processing
 					peer.Promote()
 					peer.SetIdle()
 					glog.V(logger.Detail).Infof("%s: delivered %d blocks", peer, len(blockPack.blocks))
+					if atomic.LoadInt32(&d.processing) == 0 {
+						go d.process()
+					}
 
 				case errInvalidChain:
 					// The hash chain is invalid (blocks are not ordered properly), abort
@@ -709,6 +705,71 @@ func (d *Downloader) banBlocks(peerId string, head common.Hash) error {
 	}
 }
 
+// process takes blocks from the queue and tries to import them into the chain.
+func (d *Downloader) process() (err error) {
+	// Make sure only one goroutine is ever allowed to process blocks at once
+	if !atomic.CompareAndSwapInt32(&d.processing, 0, 1) {
+		return
+	}
+	// If the processor just exited, but there are freshly pending items, try to
+	// reenter. This is needed because the goroutine spinned up for processing
+	// the fresh blocks might have been rejected entry to to this present thread
+	// not yet releasing the `processing` state.
+	defer func() {
+		if err == nil && d.queue.GetHeadBlock() != nil {
+			err = d.process()
+		}
+	}()
+	// Release the lock upon exit (note, before checking for reentry!)
+	defer atomic.StoreInt32(&d.processing, 0)
+
+	// Fetch the current cancel channel to allow termination
+	d.cancelLock.RLock()
+	cancel := d.cancelCh
+	d.cancelLock.RUnlock()
+
+	// Repeat the processing as long as there are blocks to import
+	for {
+		// Fetch the next batch of blocks
+		blocks := d.queue.TakeBlocks()
+		if len(blocks) == 0 {
+			return nil
+		}
+		// Reset the import statistics
+		d.importLock.Lock()
+		d.importStart = time.Now()
+		d.importQueue = blocks
+		d.importDone = 0
+		d.importLock.Unlock()
+
+		// Actually import the blocks
+		glog.V(logger.Debug).Infof("Inserting chain with %d blocks (#%v - #%v)\n", len(blocks), blocks[0].RawBlock.Number(), blocks[len(blocks)-1].RawBlock.Number())
+		for len(blocks) != 0 { // TODO: quit
+			// Check for any termination requests
+			select {
+			case <-cancel:
+				return errCancelChainImport
+			default:
+			}
+			// Retrieve the first batch of blocks to insert
+			max := int(math.Min(float64(len(blocks)), float64(maxBlockProcess)))
+			raw := make(types.Blocks, 0, max)
+			for _, block := range blocks[:max] {
+				raw = append(raw, block.RawBlock)
+			}
+			// Try to inset the blocks, drop the originating peer if there's an error
+			index, err := d.insertChain(raw)
+			if err != nil {
+				glog.V(logger.Debug).Infoln("Block #%d import failed:", raw[index].NumberU64(), err)
+				d.dropPeer(blocks[index].OriginPeer)
+				d.Cancel()
+				return errCancelChainImport
+			}
+			blocks = blocks[max:]
+		}
+	}
+}
+
 // DeliverBlocks injects a new batch of blocks received from a remote node.
 // This is usually invoked through the BlocksMsg by the protocol handler.
 func (d *Downloader) DeliverBlocks(id string, blocks []*types.Block) error {

eth/handler.go

@@ -80,7 +80,7 @@ func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpo
 		txsyncCh:   make(chan *txsync),
 		quitSync:   make(chan struct{}),
 	}
-	manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.removePeer)
+	manager.downloader = downloader.New(manager.eventMux, manager.chainman.HasBlock, manager.chainman.GetBlock, manager.chainman.InsertChain, manager.removePeer)
 	manager.SubProtocol = p2p.Protocol{
 		Name:    "eth",
 		Version: uint(protocolVersion),

eth/sync.go

 package eth
 
 import (
-	"math"
 	"math/rand"
-	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -15,12 +13,10 @@ import (
 
 const (
 	forceSyncCycle      = 10 * time.Second       // Time interval to force syncs, even if few peers are available
-	blockProcCycle      = 500 * time.Millisecond // Time interval to check for new blocks to process
 	notifyCheckCycle    = 100 * time.Millisecond // Time interval to allow hash notifies to fulfill before hard fetching
 	notifyArriveTimeout = 500 * time.Millisecond // Time allowance before an announced block is explicitly requested
 	notifyFetchTimeout  = 5 * time.Second        // Maximum alloted time to return an explicitly requested block
 	minDesiredPeerCount = 5                      // Amount of peers desired to start syncing
-	blockProcAmount     = 256
 
 	// This is the target size for the packs of transactions sent by txsyncLoop.
 	// A pack can get larger than this if a single transactions exceeds this size.
@@ -254,10 +250,10 @@ func (pm *ProtocolManager) fetcher() {
 // syncer is responsible for periodically synchronising with the network, both
 // downloading hashes and blocks as well as retrieving cached ones.
 func (pm *ProtocolManager) syncer() {
-	forceSync := time.Tick(forceSyncCycle)
-	blockProc := time.Tick(blockProcCycle)
-	blockProcPend := int32(0)
+	// Abort any pending syncs if we terminate
+	defer pm.downloader.Cancel()
 
+	forceSync := time.Tick(forceSyncCycle)
 	for {
 		select {
 		case <-pm.newPeerCh:
@@ -271,55 +267,12 @@ func (pm *ProtocolManager) syncer() {
 			// Force a sync even if not enough peers are present
 			go pm.synchronise(pm.peers.BestPeer())
 
-		case <-blockProc:
-			// Try to pull some blocks from the downloaded
-			if atomic.CompareAndSwapInt32(&blockProcPend, 0, 1) {
-				go func() {
-					pm.processBlocks()
-					atomic.StoreInt32(&blockProcPend, 0)
-				}()
-			}
-
 		case <-pm.quitSync:
 			return
 		}
 	}
 }
 
-// processBlocks retrieves downloaded blocks from the download cache and tries
-// to construct the local block chain with it. Note, since the block retrieval
-// order matters, access to this function *must* be synchronized/serialized.
-func (pm *ProtocolManager) processBlocks() error {
-	pm.wg.Add(1)
-	defer pm.wg.Done()
-
-	// Short circuit if no blocks are available for insertion
-	blocks := pm.downloader.TakeBlocks()
-	if len(blocks) == 0 {
-		return nil
-	}
-	glog.V(logger.Debug).Infof("Inserting chain with %d blocks (#%v - #%v)\n", len(blocks), blocks[0].RawBlock.Number(), blocks[len(blocks)-1].RawBlock.Number())
-
-	for len(blocks) != 0 && !pm.quit {
-		// Retrieve the first batch of blocks to insert
-		max := int(math.Min(float64(len(blocks)), float64(blockProcAmount)))
-		raw := make(types.Blocks, 0, max)
-		for _, block := range blocks[:max] {
-			raw = append(raw, block.RawBlock)
-		}
-		// Try to inset the blocks, drop the originating peer if there's an error
-		index, err := pm.chainman.InsertChain(raw)
-		if err != nil {
-			glog.V(logger.Debug).Infoln("Downloaded block import failed:", err)
-			pm.removePeer(blocks[index].OriginPeer)
-			pm.downloader.Cancel()
-			return err
-		}
-		blocks = blocks[max:]
-	}
-	return nil
-}
-
 // synchronise tries to sync up our local block chain with a remote peer, both
 // adding various sanity checks as well as wrapping it with various log entries.
 func (pm *ProtocolManager) synchronise(peer *peer) {