core, consensus: pluggable consensus engines (#3817)

This commit adds pluggable consensus engines to go-ethereum. In short, it
introduces a generic consensus interface, and refactors the entire codebase to
use this interface.

1 files changed, 159 insertions, 44 deletions

diff --git a/core/block_validator_test.go b/core/block_validator_test.go
index a07dd9e51..abe1766b4 100644
--- a/core/block_validator_test.go
+++ b/core/block_validator_test.go
@@ -17,64 +17,179 @@
 package core
 
 import (
-	"math/big"
+	"runtime"
 	"testing"
+	"time"
 
-	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core/state"
+	"github.com/ethereum/go-ethereum/consensus/ethash"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/params"
-	"github.com/ethereum/go-ethereum/pow"
 )
 
-func testGenesis(account common.Address, balance *big.Int) *Genesis {
-	return &Genesis{
-		Config: params.TestChainConfig,
-		Alloc:  GenesisAlloc{account: {Balance: balance}},
+// Tests that simple header verification works, for both good and bad blocks.
+func TestHeaderVerification(t *testing.T) {
+	// Create a simple chain to verify
+	var (
+		testdb, _ = ethdb.NewMemDatabase()
+		gspec     = &Genesis{Config: params.TestChainConfig}
+		genesis   = gspec.MustCommit(testdb)
+		blocks, _ = GenerateChain(params.TestChainConfig, genesis, testdb, 8, nil)
+	)
+	headers := make([]*types.Header, len(blocks))
+	for i, block := range blocks {
+		headers[i] = block.Header()
+	}
+	// Run the header checker for blocks one-by-one, checking for both valid and invalid nonces
+	chain, _ := NewBlockChain(testdb, params.TestChainConfig, ethash.NewFaker(), new(event.TypeMux), vm.Config{})
+
+	for i := 0; i < len(blocks); i++ {
+		for j, valid := range []bool{true, false} {
+			var results <-chan error
+
+			if valid {
+				engine := ethash.NewFaker()
+				_, results = engine.VerifyHeaders(chain, []*types.Header{headers[i]}, []bool{true})
+			} else {
+				engine := ethash.NewFakeFailer(headers[i].Number.Uint64())
+				_, results = engine.VerifyHeaders(chain, []*types.Header{headers[i]}, []bool{true})
+			}
+			// Wait for the verification result
+			select {
+			case result := <-results:
+				if (result == nil) != valid {
+					t.Errorf("test %d.%d: validity mismatch: have %v, want %v", i, j, result, valid)
+				}
+			case <-time.After(time.Second):
+				t.Fatalf("test %d.%d: verification timeout", i, j)
+			}
+			// Make sure no more data is returned
+			select {
+			case result := <-results:
+				t.Fatalf("test %d.%d: unexpected result returned: %v", i, j, result)
+			case <-time.After(25 * time.Millisecond):
+			}
+		}
+		chain.InsertChain(blocks[i : i+1])
 	}
 }
 
-func TestNumber(t *testing.T) {
-	chain := newTestBlockChain()
-	statedb, _ := state.New(chain.Genesis().Root(), chain.chainDb)
-	header := makeHeader(chain.config, chain.Genesis(), statedb)
-	header.Number = big.NewInt(3)
-	err := ValidateHeader(chain.config, pow.FakePow{}, header, chain.Genesis().Header(), false, false)
-	if err != BlockNumberErr {
-		t.Errorf("expected block number error, got %q", err)
+// Tests that concurrent header verification works, for both good and bad blocks.
+func TestHeaderConcurrentVerification2(t *testing.T)  { testHeaderConcurrentVerification(t, 2) }
+func TestHeaderConcurrentVerification8(t *testing.T)  { testHeaderConcurrentVerification(t, 8) }
+func TestHeaderConcurrentVerification32(t *testing.T) { testHeaderConcurrentVerification(t, 32) }
+
+func testHeaderConcurrentVerification(t *testing.T, threads int) {
+	// Create a simple chain to verify
+	var (
+		testdb, _ = ethdb.NewMemDatabase()
+		gspec     = &Genesis{Config: params.TestChainConfig}
+		genesis   = gspec.MustCommit(testdb)
+		blocks, _ = GenerateChain(params.TestChainConfig, genesis, testdb, 8, nil)
+	)
+	headers := make([]*types.Header, len(blocks))
+	seals := make([]bool, len(blocks))
+
+	for i, block := range blocks {
+		headers[i] = block.Header()
+		seals[i] = true
 	}
+	// Set the number of threads to verify on
+	old := runtime.GOMAXPROCS(threads)
+	defer runtime.GOMAXPROCS(old)
+
+	// Run the header checker for the entire block chain at once both for a valid and
+	// also an invalid chain (enough if one arbitrary block is invalid).
+	for i, valid := range []bool{true, false} {
+		var results <-chan error
+
+		if valid {
+			chain, _ := NewBlockChain(testdb, params.TestChainConfig, ethash.NewFaker(), new(event.TypeMux), vm.Config{})
+			_, results = chain.engine.VerifyHeaders(chain, headers, seals)
+		} else {
+			chain, _ := NewBlockChain(testdb, params.TestChainConfig, ethash.NewFakeFailer(uint64(len(headers)-1)), new(event.TypeMux), vm.Config{})
+			_, results = chain.engine.VerifyHeaders(chain, headers, seals)
+		}
+		// Wait for all the verification results
+		checks := make(map[int]error)
+		for j := 0; j < len(blocks); j++ {
+			select {
+			case result := <-results:
+				checks[j] = result
 
-	header = makeHeader(chain.config, chain.Genesis(), statedb)
-	err = ValidateHeader(chain.config, pow.FakePow{}, header, chain.Genesis().Header(), false, false)
-	if err == BlockNumberErr {
-		t.Errorf("didn't expect block number error")
+			case <-time.After(time.Second):
+				t.Fatalf("test %d.%d: verification timeout", i, j)
+			}
+		}
+		// Check nonce check validity
+		for j := 0; j < len(blocks); j++ {
+			want := valid || (j < len(blocks)-2) // We chose the last-but-one nonce in the chain to fail
+			if (checks[j] == nil) != want {
+				t.Errorf("test %d.%d: validity mismatch: have %v, want %v", i, j, checks[j], want)
+			}
+			if !want {
+				// A few blocks after the first error may pass verification due to concurrent
+				// workers. We don't care about those in this test, just that the correct block
+				// errors out.
+				break
+			}
+		}
+		// Make sure no more data is returned
+		select {
+		case result := <-results:
+			t.Fatalf("test %d: unexpected result returned: %v", i, result)
+		case <-time.After(25 * time.Millisecond):
+		}
 	}
 }
 
-func TestPutReceipt(t *testing.T) {
-	db, _ := ethdb.NewMemDatabase()
-
-	var addr common.Address
-	addr[0] = 1
-	var hash common.Hash
-	hash[0] = 2
-
-	receipt := new(types.Receipt)
-	receipt.Logs = []*types.Log{{
-		Address:     addr,
-		Topics:      []common.Hash{hash},
-		Data:        []byte("hi"),
-		BlockNumber: 42,
-		TxHash:      hash,
-		TxIndex:     0,
-		BlockHash:   hash,
-		Index:       0,
-	}}
-
-	WriteReceipts(db, types.Receipts{receipt})
-	receipt = GetReceipt(db, common.Hash{})
-	if receipt == nil {
-		t.Error("expected to get 1 receipt, got none.")
+// Tests that aborting a header validation indeed prevents further checks from being
+// run, as well as checks that no left-over goroutines are leaked.
+func TestHeaderConcurrentAbortion2(t *testing.T)  { testHeaderConcurrentAbortion(t, 2) }
+func TestHeaderConcurrentAbortion8(t *testing.T)  { testHeaderConcurrentAbortion(t, 8) }
+func TestHeaderConcurrentAbortion32(t *testing.T) { testHeaderConcurrentAbortion(t, 32) }
+
+func testHeaderConcurrentAbortion(t *testing.T, threads int) {
+	// Create a simple chain to verify
+	var (
+		testdb, _ = ethdb.NewMemDatabase()
+		gspec     = &Genesis{Config: params.TestChainConfig}
+		genesis   = gspec.MustCommit(testdb)
+		blocks, _ = GenerateChain(params.TestChainConfig, genesis, testdb, 1024, nil)
+	)
+	headers := make([]*types.Header, len(blocks))
+	seals := make([]bool, len(blocks))
+
+	for i, block := range blocks {
+		headers[i] = block.Header()
+		seals[i] = true
+	}
+	// Set the number of threads to verify on
+	old := runtime.GOMAXPROCS(threads)
+	defer runtime.GOMAXPROCS(old)
+
+	// Start the verifications and immediately abort
+	chain, _ := NewBlockChain(testdb, params.TestChainConfig, ethash.NewFakeDelayer(time.Millisecond), new(event.TypeMux), vm.Config{})
+	abort, results := chain.engine.VerifyHeaders(chain, headers, seals)
+	close(abort)
+
+	// Deplete the results channel
+	verified := 0
+	for depleted := false; !depleted; {
+		select {
+		case result := <-results:
+			if result != nil {
+				t.Errorf("header %d: validation failed: %v", verified, result)
+			}
+			verified++
+		case <-time.After(50 * time.Millisecond):
+			depleted = true
+		}
+	}
+	// Check that abortion was honored by not processing too many POWs
+	if verified > 2*threads {
+		t.Errorf("verification count too large: have %d, want below %d", verified, 2*threads)
 	}
 }