Merge pull request ethereum#1822 from karalabe/contain-pow

core: separate and contain POW verifier, extensive tests

Author: obscuren

core/chain_manager.go

@@ -22,7 +22,6 @@ import (
 	"fmt"
 	"io"
 	"math/big"
-	"runtime"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -616,14 +615,12 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
 		stats      struct{ queued, processed, ignored int }
 		tstart     = time.Now()
 
-		nonceDone    = make(chan nonceResult, len(chain))
-		nonceQuit    = make(chan struct{})
 		nonceChecked = make([]bool, len(chain))
 	)
 
 	// Start the parallel nonce verifier.
-	go verifyNonces(self.pow, chain, nonceQuit, nonceDone)
-	defer close(nonceQuit)
+	nonceAbort, nonceResults := verifyNoncesFromBlocks(self.pow, chain)
+	defer close(nonceAbort)
 
 	txcount := 0
 	for i, block := range chain {
@@ -636,11 +633,11 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
 		// Wait for block i's nonce to be verified before processing
 		// its state transition.
 		for !nonceChecked[i] {
-			r := <-nonceDone
-			nonceChecked[r.i] = true
+			r := <-nonceResults
+			nonceChecked[r.index] = true
 			if !r.valid {
-				block := chain[r.i]
-				return r.i, &BlockNonceErr{Hash: block.Hash(), Number: block.Number(), Nonce: block.Nonce()}
+				block := chain[r.index]
+				return r.index, &BlockNonceErr{Hash: block.Hash(), Number: block.Number(), Nonce: block.Nonce()}
 			}
 		}
 
@@ -843,40 +840,3 @@ func blockErr(block *types.Block, err error) {
 	glog.V(logger.Error).Infoln(err)
 	glog.V(logger.Debug).Infoln(verifyNonces)
 }
-
-type nonceResult struct {
-	i     int
-	valid bool
-}
-
-// block verifies nonces of the given blocks in parallel and returns
-// an error if one of the blocks nonce verifications failed.
-func verifyNonces(pow pow.PoW, blocks []*types.Block, quit <-chan struct{}, done chan<- nonceResult) {
-	// Spawn a few workers. They listen for blocks on the in channel
-	// and send results on done. The workers will exit in the
-	// background when in is closed.
-	var (
-		in       = make(chan int)
-		nworkers = runtime.GOMAXPROCS(0)
-	)
-	defer close(in)
-	if len(blocks) < nworkers {
-		nworkers = len(blocks)
-	}
-	for i := 0; i < nworkers; i++ {
-		go func() {
-			for i := range in {
-				done <- nonceResult{i: i, valid: pow.Verify(blocks[i])}
-			}
-		}()
-	}
-	// Feed block indices to the workers.
-	for i := range blocks {
-		select {
-		case in <- i:
-			continue
-		case <-quit:
-			return
-		}
-	}
-}

core/chain_manager_test.go

@@ -456,7 +456,7 @@ func TestInsertNonceError(t *testing.T) {
 
 		fail := rand.Int() % len(blocks)
 		failblock := blocks[fail]
-		bc.pow = failpow{failblock.NumberU64()}
+		bc.pow = failPow{failblock.NumberU64()}
 		n, err := bc.InsertChain(blocks)
 
 		// Check that the returned error indicates the nonce failure.
@@ -499,18 +499,3 @@ func TestGenesisMismatch(t *testing.T) {
 	}
 }
 */
-
-// failpow returns false from Verify for a certain block number.
-type failpow struct{ num uint64 }
-
-func (pow failpow) Search(pow.Block, <-chan struct{}) (nonce uint64, mixHash []byte) {
-	return 0, nil
-}
-func (pow failpow) Verify(b pow.Block) bool {
-	return b.NumberU64() != pow.num
-}
-func (pow failpow) GetHashrate() int64 {
-	return 0
-}
-func (pow failpow) Turbo(bool) {
-}

core/chain_pow.go

@@ -0,0 +1,87 @@
+// Copyright 2015 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package core
+
+import (
+	"runtime"
+
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/pow"
+)
+
+// nonceCheckResult contains the result of a nonce verification.
+type nonceCheckResult struct {
+	index int  // Index of the item verified from an input array
+	valid bool // Result of the nonce verification
+}
+
+// verifyNoncesFromHeaders starts a concurrent header nonce verification,
+// returning a quit channel to abort the operations and a results channel
+// to retrieve the async verifications.
+func verifyNoncesFromHeaders(checker pow.PoW, headers []*types.Header) (chan<- struct{}, <-chan nonceCheckResult) {
+	items := make([]pow.Block, len(headers))
+	for i, header := range headers {
+		items[i] = types.NewBlockWithHeader(header)
+	}
+	return verifyNonces(checker, items)
+}
+
+// verifyNoncesFromBlocks starts a concurrent block nonce verification,
+// returning a quit channel to abort the operations and a results channel
+// to retrieve the async verifications.
+func verifyNoncesFromBlocks(checker pow.PoW, blocks []*types.Block) (chan<- struct{}, <-chan nonceCheckResult) {
+	items := make([]pow.Block, len(blocks))
+	for i, block := range blocks {
+		items[i] = block
+	}
+	return verifyNonces(checker, items)
+}
+
+// verifyNonces starts a concurrent nonce verification, returning a quit channel
+// to abort the operations and a results channel to retrieve the async checks.
+func verifyNonces(checker pow.PoW, items []pow.Block) (chan<- struct{}, <-chan nonceCheckResult) {
+	// Spawn as many workers as allowed threads
+	workers := runtime.GOMAXPROCS(0)
+	if len(items) < workers {
+		workers = len(items)
+	}
+	// Create a task channel and spawn the verifiers
+	tasks := make(chan int, workers)
+	results := make(chan nonceCheckResult, len(items)) // Buffered to make sure all workers stop
+	for i := 0; i < workers; i++ {
+		go func() {
+			for index := range tasks {
+				results <- nonceCheckResult{index: index, valid: checker.Verify(items[index])}
+			}
+		}()
+	}
+	// Feed item indices to the workers until done or aborted
+	abort := make(chan struct{})
+	go func() {
+		defer close(tasks)
+
+		for i := range items {
+			select {
+			case tasks <- i:
+				continue
+			case <-abort:
+				return
+			}
+		}
+	}()
+	return abort, results
+}