Make cmd/tester initiate staged sync, remove some pools and caches (#746

)

* no pools

* Remove excessive logging

* Revert

* Initial commit

* Cleanup

* Fix some compile errors

* Fix core tests

* Fix more tests

* Fix eth/downloader

* go mod tidy

* Fix core/vm

* Fix formatting

* Fix formatting

* Fix lint

* Fix lint

* Switch tests to StagedSync

* Cleanup

* Reuse cache in stage4

* Fix

* Fix formatting

* Try to fix test

cmd/geth/chaincmd.go

@@ -246,7 +246,7 @@ func initGenesis(ctx *cli.Context) error {
 		if err != nil {
 			utils.Fatalf("Failed to open database: %v", err)
 		}
-		_, hash, _, err := core.SetupGenesisBlock(chaindb, genesis, false /* history */)
+		_, hash, _, err := core.SetupGenesisBlock(chaindb, genesis, false /* history */, false /* overwrite */)
 		if err != nil {
 			utils.Fatalf("Failed to write genesis block: %v", err)
 		}
@@ -278,7 +278,7 @@ func importChain(ctx *cli.Context) error {
 	stack := makeFullNode(ctx)
 	defer stack.Close()
 
-	chain, db := utils.MakeChain(ctx, stack, false)
+	_, chain, db := utils.MakeChain(ctx, stack, false)
 	defer db.Close()
 
 	// Start periodically gathering memory profiles
@@ -332,7 +332,7 @@ func exportChain(ctx *cli.Context) error {
 	stack := makeFullNode(ctx)
 	defer stack.Close()
 
-	chain, _ := utils.MakeChain(ctx, stack, true)
+	_, chain, _ := utils.MakeChain(ctx, stack, true)
 	start := time.Now()
 
 	var err error
@@ -408,14 +408,14 @@ func copyDb(ctx *cli.Context) error {
 	stack := makeFullNode(ctx)
 	defer stack.Close()
 
-	chain, chainDb := utils.MakeChain(ctx, stack, false)
+	chainConfig, chain, chainDb := utils.MakeChain(ctx, stack, false)
 	syncMode := *utils.GlobalTextMarshaler(ctx, utils.SyncModeFlag.Name).(*downloader.SyncMode)
 
 	var syncBloom *trie.SyncBloom
 	if syncMode == downloader.FastSync {
 		//syncBloom = trie.NewSyncBloom(uint64(ctx.GlobalInt(utils.CacheFlag.Name)/2), chainDb)
 	}
-	dl := downloader.New(0, chainDb, syncBloom, new(event.TypeMux), chain, nil, nil, ethdb.DefaultStorageMode)
+	dl := downloader.New(0, chainDb, syncBloom, new(event.TypeMux), chainConfig, chain, nil, nil, ethdb.DefaultStorageMode)
 
 	// Create a source peer to satisfy downloader requests from
 	db := ethdb.MustOpen(ctx.Args().First())
@@ -506,7 +506,7 @@ func dump(ctx *cli.Context) error {
 	stack := makeFullNode(ctx)
 	defer stack.Close()
 
-	chain, chainDb := utils.MakeChain(ctx, stack, true)
+	_, chain, chainDb := utils.MakeChain(ctx, stack, true)
 	defer chainDb.Close()
 	for _, arg := range ctx.Args() {
 		var block *types.Block
@@ -536,7 +536,7 @@ func inspect(ctx *cli.Context) error {
 	node, _ := makeConfigNode(ctx)
 	defer node.Close()
 
-	_, chainDb := utils.MakeChain(ctx, node, true)
+	_, _, chainDb := utils.MakeChain(ctx, node, true)
 	defer chainDb.Close()
 
 	return ethdb.InspectDatabase(chainDb)

cmd/geth/retesteth.go

@@ -381,7 +381,7 @@ func (api *RetestethAPI) SetChainParams(_ context.Context, chainParams ChainPara
 		ParentHash: chainParams.Genesis.ParentHash,
 		Alloc:      accounts,
 	}
-	chainConfig, genesisHash, _, err := core.SetupGenesisBlock(ethDb, genesis, false /* history */)
+	chainConfig, genesisHash, _, err := core.SetupGenesisBlock(ethDb, genesis, false /* history */, false /* overwrite */)
 	if err != nil {
 		return false, err
 	}

cmd/hack/hack.go

@@ -611,7 +611,7 @@ func execToBlock(chaindata string, block uint64, fromScratch bool) {
 	defer stateDB.Close()
 
 	//_, _, _, err = core.SetupGenesisBlock(stateDB, core.DefaultGenesisBlock())
-	_, _, _, err = core.SetupGenesisBlock(stateDB, nil, false /* history */)
+	_, _, _, err = core.SetupGenesisBlock(stateDB, nil, false /* history */, true /* overwrite */)
 	check(err)
 	bc, err := core.NewBlockChain(stateDB, nil, params.MainnetChainConfig, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
 	check(err)
@@ -1852,9 +1852,31 @@ func fixStages(chaindata string) error {
 }
 
 func changeSetStats(chaindata string, block1, block2 uint64) error {
-	fmt.Printf("Changeset stats from %d to %d\n", block1, block2)
 	db := ethdb.MustOpen(chaindata)
 	defer db.Close()
+	fmt.Printf("State stats\n")
+	stAccounts := 0
+	stStorage := 0
+	if err := db.KV().View(context.Background(), func(tx ethdb.Tx) error {
+		st := tx.Bucket(dbutils.PlainStateBucket)
+		c := st.Cursor()
+		k, _, e := c.First()
+		for ; k != nil && e == nil; k, _, e = c.Next() {
+			if len(k) > 28 {
+				stStorage++
+			} else {
+				stAccounts++
+			}
+			if (stStorage+stAccounts)%100000 == 0 {
+				fmt.Printf("State records: %d\n", stStorage+stAccounts)
+			}
+		}
+		return e
+	}); err != nil {
+		return err
+	}
+	fmt.Printf("stAccounts = %d, stStorage = %d\n", stAccounts, stStorage)
+	fmt.Printf("Changeset stats from %d to %d\n", block1, block2)
 	accounts := make(map[string]struct{})
 	if err := db.KV().View(context.Background(), func(tx ethdb.Tx) error {
 		st := tx.Bucket(dbutils.PlainAccountChangeSetBucket)
@@ -1868,7 +1890,7 @@ func changeSetStats(chaindata string, block1, block2 uint64) error {
 			if timestamp >= block2 {
 				break
 			}
-			if timestamp%100000 == 0 {
+			if (timestamp-block1)%100000 == 0 {
 				fmt.Printf("at the block %d for accounts, booster size: %d\n", timestamp, len(accounts))
 			}
 			if err1 := changeset.AccountChangeSetPlainBytes(v).Walk(func(kk, _ []byte) error {
@@ -1895,7 +1917,7 @@ func changeSetStats(chaindata string, block1, block2 uint64) error {
 			if timestamp >= block2 {
 				break
 			}
-			if timestamp%100000 == 0 {
+			if (timestamp-block1)%100000 == 0 {
 				fmt.Printf("at the block %d for storage, booster size: %d\n", timestamp, len(storage))
 			}
 			if err1 := changeset.StorageChangeSetPlainBytes(v).Walk(func(kk, _ []byte) error {

cmd/integration/commands/stages.go

@@ -198,7 +198,7 @@ func stage4(ctx context.Context) error {
 		// TODO
 	}
 
-	blockchain, err := newBlockChain(db)
+	chainConfig, blockchain, err := newBlockChain(db)
 	if err != nil {
 		return err
 	}
@@ -209,7 +209,7 @@ func stage4(ctx context.Context) error {
 		u := &stagedsync.UnwindState{Stage: stages.Execution, UnwindPoint: stage4.BlockNumber - unwind}
 		return stagedsync.UnwindExecutionStage(u, stage4, db)
 	}
-	return stagedsync.SpawnExecuteBlocksStage(stage4, db, blockchain, block, ch, blockchain.DestsCache, false, nil)
+	return stagedsync.SpawnExecuteBlocksStage(stage4, db, chainConfig, blockchain, block, ch, blockchain.DestsCache, false, nil)
 }
 
 func stage5(ctx context.Context) error {
@@ -324,12 +324,16 @@ func printAllStages(_ context.Context) error {
 	return printStages(db)
 }
 
-func newBlockChain(db ethdb.Database) (*core.BlockChain, error) {
+func newBlockChain(db ethdb.Database) (*params.ChainConfig, *core.BlockChain, error) {
 	config := eth.DefaultConfig
-	chainConfig, _, _, err := core.SetupGenesisBlock(db, config.Genesis, config.StorageMode.History)
+	chainConfig, _, _, err := core.SetupGenesisBlock(db, config.Genesis, config.StorageMode.History, true /* overwrite */)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	vmConfig, cacheConfig, dests := eth.BlockchainRuntimeConfig(&config)
-	return core.NewBlockChain(db, cacheConfig, chainConfig, ethash.NewFaker(), vmConfig, nil, &config.TxLookupLimit, dests)
+	blockchain, err1 := core.NewBlockChain(db, cacheConfig, chainConfig, ethash.NewFaker(), vmConfig, nil, &config.TxLookupLimit, dests)
+	if err1 != nil {
+		return nil, nil, err1
+	}
+	return chainConfig, blockchain, nil
 }

cmd/integration/commands/state_stages.go

@@ -57,7 +57,7 @@ func syncBySmallSteps(ctx context.Context, chaindata string) error {
 	core.UsePlainStateExecution = true
 	db := ethdb.MustOpen(chaindata)
 	defer db.Close()
-	blockchain, chainErr := newBlockChain(db)
+	chainConfig, blockchain, chainErr := newBlockChain(db)
 	if chainErr != nil {
 		return chainErr
 	}
@@ -111,7 +111,7 @@ func syncBySmallSteps(ctx context.Context, chaindata string) error {
 				unwind = 0
 			}
 
-			if err := stagedsync.SpawnExecuteBlocksStage(stage, db, blockchain, execToBlock, ch, nil, false, changeSetHook); err != nil {
+			if err := stagedsync.SpawnExecuteBlocksStage(stage, db, chainConfig, blockchain, execToBlock, ch, nil, false, changeSetHook); err != nil {
 				return fmt.Errorf("spawnExecuteBlocksStage: %w", err)
 			}
 		}

cmd/state/stateless/stateless.go

@@ -190,7 +190,7 @@ func Stateless(
 	}
 	var preRoot common.Hash
 	if blockNum == 1 {
-		_, _, _, err = core.SetupGenesisBlock(stateDb, core.DefaultGenesisBlock(), writeHistory)
+		_, _, _, err = core.SetupGenesisBlock(stateDb, core.DefaultGenesisBlock(), writeHistory, true /* overwrite */)
 		check(err)
 		genesisBlock, _, _, err1 := core.DefaultGenesisBlock().ToBlock(nil, writeHistory)
 		check(err1)

cmd/tester/block_generator.go

@@ -17,10 +17,8 @@ import (
 	"github.com/ledgerwatch/turbo-geth/common"
 	"github.com/ledgerwatch/turbo-geth/consensus/ethash"
 	"github.com/ledgerwatch/turbo-geth/core"
-	"github.com/ledgerwatch/turbo-geth/core/forkid"
 	"github.com/ledgerwatch/turbo-geth/core/rawdb"
 	"github.com/ledgerwatch/turbo-geth/core/types"
-	"github.com/ledgerwatch/turbo-geth/core/vm"
 	"github.com/ledgerwatch/turbo-geth/crypto"
 	"github.com/ledgerwatch/turbo-geth/ethdb"
 	"github.com/ledgerwatch/turbo-geth/log"
@@ -41,7 +39,6 @@ type BlockGenerator struct {
 	tdByNumber          map[uint64]*big.Int
 	lastBlock           *types.Block
 	totalDifficulty     *big.Int
-	forkId              forkid.ID
 }
 
 func (bg *BlockGenerator) Close() {
@@ -100,10 +97,6 @@ func (bg *BlockGenerator) LastBlock() *types.Block {
 	return bg.lastBlock
 }
 
-func (bg *BlockGenerator) ForkID() forkid.ID {
-	return bg.forkId
-}
-
 func randAddress(r *rand.Rand) common.Address {
 	if ReduceComplexity {
 		return common.BytesToAddress(common.FromHex("80977316944e5942e79b0e3abad38da746086519"))
@@ -367,12 +360,6 @@ func NewBlockGenerator(ctx context.Context, outputFile string, initialHeight int
 		return nil, err
 	}
 
-	blockchain, err := core.NewBlockChain(db, nil, genesis.Config, ethash.NewFullFaker(), vm.Config{}, nil, nil, nil)
-	if err != nil {
-		return nil, err
-	}
-	bg.forkId = forkid.NewID(blockchain)
-
 	bg.input, err = os.Open(outputFile)
 	// Reopen the file for reading
 	if err != nil {
@@ -428,12 +415,6 @@ func NewForkGenerator(ctx context.Context, base *BlockGenerator, outputFile stri
 		return nil, err
 	}
 
-	blockchain, err := core.NewBlockChain(db, nil, genesis.Config, ethash.NewFullFaker(), vm.Config{}, nil, nil, nil)
-	if err != nil {
-		return nil, err
-	}
-	bg.forkId = forkid.NewID(blockchain)
-
 	bg.input, err = os.Open(outputFile)
 	if err != nil {
 		return nil, err

cmd/tester/interfaces.go

@@ -4,7 +4,6 @@ import (
 	"math/big"
 
 	"github.com/ledgerwatch/turbo-geth/common"
-	"github.com/ledgerwatch/turbo-geth/core/forkid"
 	"github.com/ledgerwatch/turbo-geth/core/types"
 )
 
@@ -17,6 +16,5 @@ type BlockFeeder interface {
 	GetTdByNumber(number uint64) *big.Int
 	TotalDifficulty() *big.Int
 	LastBlock() *types.Block
-	ForkID() forkid.ID
 	Genesis() *types.Block
 }

cmd/tester/main.go

@@ -12,7 +12,6 @@ import (
 	"runtime"
 	"sort"
 	"syscall"
-	"time"
 
 	"github.com/ledgerwatch/turbo-geth/cmd/utils"
 	"github.com/ledgerwatch/turbo-geth/console/prompt"
@@ -129,24 +128,16 @@ func tester(cliCtx *cli.Context) error {
 	}
 
 	tp := NewTesterProtocol()
-	server := makeP2PServer(ctx, tp, []string{eth.DebugName})
-	// Add protocol
-	if err := server.Start(); err != nil {
-		panic(fmt.Errorf("could not start server: %w", err))
-	}
-	server.AddPeer(nodeToConnect)
-	time.Sleep(time.Second)
-	server.Stop()
 
 	//fmt.Printf("%s %s\n", ctx.Args()[0], ctx.Args()[1])
 	//tp.blockFeeder, err = NewBlockAccessor(ctx.Args()[0]/*, ctx.Args()[1]*/)
-	blockGen, err := NewBlockGenerator(ctx, "blocks", 50000)
+	blockGen, err := NewBlockGenerator(ctx, "blocks", 5)
 	if err != nil {
 		panic(fmt.Sprintf("Failed to create block generator: %v", err))
 	}
 	defer blockGen.Close()
 	tp.blockFeeder = blockGen
-	tp.forkBase = 49998
+	tp.forkBase = 3
 	tp.forkHeight = 5
 	tp.forkFeeder, err = NewForkGenerator(ctx, blockGen, "forkblocks", tp.forkBase, tp.forkHeight)
 	if err != nil {
@@ -157,7 +148,7 @@ func tester(cliCtx *cli.Context) error {
 	tp.networkId = 1 // Mainnet
 	tp.genesisBlockHash = tp.forkFeeder.Genesis().Hash()
 
-	server = makeP2PServer(ctx, tp, []string{eth.ProtocolName})
+	server := makeP2PServer(ctx, tp, []string{eth.ProtocolName, eth.DebugName})
 	// Add protocol
 	if err := server.Start(); err != nil {
 		panic(fmt.Errorf("could not start server: %w", err))