forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

cmd, core, eth: version 2

Browse source
This commit is contained in:
rjl493456442 2019-07-19 15:43:08 +08:00 committed by Péter Szilágyi
parent 4f0cc2d7e2
commit 2c7b166b25
No known key found for this signature in database
GPG key ID: E9AE538CEDF8293D
30 changed files with 649 additions and 266 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
accounts/abi/bind/backends/simulated.go
View file

@ -70,7 +70,7 @@ type SimulatedBackend struct {
func NewSimulatedBackendWithDatabase(database ethdb.Database, alloc core.GenesisAlloc, gasLimit uint64) *SimulatedBackend {
genesis := core.Genesis{Config: params.AllEthashProtocolChanges, GasLimit: gasLimit, Alloc: alloc}
genesis.MustCommit(database)
blockchain, _ := core.NewBlockChain(database, nil, genesis.Config, ethash.NewFaker(), vm.Config{}, nil)
blockchain, _ := core.NewBlockChain(database, nil, genesis.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
backend := &SimulatedBackend{
database: database,

1
cmd/geth/main.go
View file

@ -94,6 +94,7 @@ var (
utils.SyncModeFlag,
utils.ExitWhenSyncedFlag,
utils.GCModeFlag,
utils.TxLookupLimitFlag,
utils.LightServeFlag,
utils.LightLegacyServFlag,
utils.LightIngressFlag,

2
cmd/geth/retesteth.go
View file

@ -398,7 +398,7 @@ func (api *RetestethAPI) SetChainParams(ctx context.Context, chainParams ChainPa
}
engine := &NoRewardEngine{inner: inner, rewardsOn: chainParams.SealEngine != "NoReward"}
blockchain, err := core.NewBlockChain(ethDb, nil, chainConfig, engine, vm.Config{}, nil)
blockchain, err := core.NewBlockChain(ethDb, nil, chainConfig, engine, vm.Config{}, nil, 0)
if err != nil {
return false, err
}

1
cmd/geth/usage.go
View file

@ -80,6 +80,7 @@ var AppHelpFlagGroups = []flagGroup{
utils.SyncModeFlag,
utils.ExitWhenSyncedFlag,
utils.GCModeFlag,
utils.TxLookupLimitFlag,
utils.EthStatsURLFlag,
utils.IdentityFlag,
utils.LightKDFFlag,

11
cmd/utils/flags.go
View file

@ -215,6 +215,11 @@ var (
Usage: `Blockchain garbage collection mode ("full", "archive")`,
Value: "full",
}
TxLookupLimitFlag = cli.Int64Flag{
Name: "txlookuplimit",
Usage: "The maximum number of blocks from head whose tx indices are reserved(0 means reserve all indices)",
Value: 0,
}
LightKDFFlag = cli.BoolFlag{
Name: "lightkdf",
Usage: "Reduce key-derivation RAM & CPU usage at some expense of KDF strength",
@ -1450,7 +1455,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) {
}
cfg.NoPruning = ctx.GlobalString(GCModeFlag.Name) == "archive"
cfg.NoPrefetch = ctx.GlobalBool(CacheNoPrefetchFlag.Name)
if ctx.GlobalIsSet(TxLookupLimitFlag.Name) {
cfg.TxLookupLimit = ctx.GlobalUint64(TxLookupLimitFlag.Name)
}
if ctx.GlobalIsSet(CacheFlag.Name) || ctx.GlobalIsSet(CacheTrieFlag.Name) {
cfg.TrieCleanCache = ctx.GlobalInt(CacheFlag.Name) * ctx.GlobalInt(CacheTrieFlag.Name) / 100
}
@ -1716,7 +1723,7 @@ func MakeChain(ctx *cli.Context, stack *node.Node) (chain *core.BlockChain, chai
cache.TrieDirtyLimit = ctx.GlobalInt(CacheFlag.Name) * ctx.GlobalInt(CacheGCFlag.Name) / 100
}
vmcfg := vm.Config{EnablePreimageRecording: ctx.GlobalBool(VMEnableDebugFlag.Name)}
chain, err = core.NewBlockChain(chainDb, cache, config, engine, vmcfg, nil)
chain, err = core.NewBlockChain(chainDb, cache, config, engine, vmcfg, nil, 0)
if err != nil {
Fatalf("Can't create BlockChain: %v", err)
}

6
consensus/clique/clique_test.go
View file

@ -54,7 +54,7 @@ func TestReimportMirroredState(t *testing.T) {
genesis := genspec.MustCommit(db)
// Generate a batch of blocks, each properly signed
chain, _ := core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil)
chain, _ := core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil, 0)
defer chain.Stop()
blocks, _ := core.GenerateChain(params.AllCliqueProtocolChanges, genesis, engine, db, 3, func(i int, block *core.BlockGen) {
@ -88,7 +88,7 @@ func TestReimportMirroredState(t *testing.T) {
db = rawdb.NewMemoryDatabase()
genspec.MustCommit(db)
chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil)
chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil, 0)
defer chain.Stop()
if _, err := chain.InsertChain(blocks[:2]); err != nil {
@ -101,7 +101,7 @@ func TestReimportMirroredState(t *testing.T) {
// Simulate a crash by creating a new chain on top of the database, without
// flushing the dirty states out. Insert the last block, trigerring a sidechain
// reimport.
chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil)
chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil, 0)
defer chain.Stop()
if _, err := chain.InsertChain(blocks[2:]); err != nil {

2
consensus/clique/snapshot_test.go
View file

@ -448,7 +448,7 @@ func TestClique(t *testing.T) {
batches[len(batches)-1] = append(batches[len(batches)-1], block)
}
// Pass all the headers through clique and ensure tallying succeeds
chain, err := core.NewBlockChain(db, nil, &config, engine, vm.Config{}, nil)
chain, err := core.NewBlockChain(db, nil, &config, engine, vm.Config{}, nil, 0)
if err != nil {
t.Errorf("test %d: failed to create test chain: %v", i, err)
continue

4
core/bench_test.go
View file

@ -175,7 +175,7 @@ func benchInsertChain(b *testing.B, disk bool, gen func(int, *BlockGen)) {
// Time the insertion of the new chain.
// State and blocks are stored in the same DB.
chainman, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
chainman, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer chainman.Stop()
b.ReportAllocs()
b.ResetTimer()
@ -287,7 +287,7 @@ func benchReadChain(b *testing.B, full bool, count uint64) {
if err != nil {
b.Fatalf("error opening database at %v: %v", dir, err)
}
chain, err := NewBlockChain(db, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil)
chain, err := NewBlockChain(db, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0)
if err != nil {
b.Fatalf("error creating chain: %v", err)
}

8
core/block_validator_test.go
View file

@ -42,7 +42,7 @@ func TestHeaderVerification(t *testing.T) {
headers[i] = block.Header()
}
// Run the header checker for blocks one-by-one, checking for both valid and invalid nonces
chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil)
chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0)
defer chain.Stop()
for i := 0; i < len(blocks); i++ {
@ -106,11 +106,11 @@ func testHeaderConcurrentVerification(t *testing.T, threads int) {
var results <-chan error
if valid {
chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil)
chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0)
_, results = chain.engine.VerifyHeaders(chain, headers, seals)
chain.Stop()
} else {
chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeFailer(uint64(len(headers)-1)), vm.Config{}, nil)
chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeFailer(uint64(len(headers)-1)), vm.Config{}, nil, 0)
_, results = chain.engine.VerifyHeaders(chain, headers, seals)
chain.Stop()
}
@ -173,7 +173,7 @@ func testHeaderConcurrentAbortion(t *testing.T, threads int) {
defer runtime.GOMAXPROCS(old)
// Start the verifications and immediately abort
chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeDelayer(time.Millisecond), vm.Config{}, nil)
chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeDelayer(time.Millisecond), vm.Config{}, nil, 0)
defer chain.Stop()
abort, results := chain.engine.VerifyHeaders(chain, headers, seals)

112
core/blockchain.go
View file

@ -134,9 +134,10 @@ type BlockChain struct {
chainConfig *params.ChainConfig // Chain & network configuration
cacheConfig *CacheConfig // Cache configuration for pruning
db ethdb.Database // Low level persistent database to store final content in
triegc *prque.Prque // Priority queue mapping block numbers to tries to gc
gcproc time.Duration // Accumulates canonical block processing for trie dumping
db ethdb.Database // Low level persistent database to store final content in
triegc *prque.Prque // Priority queue mapping block numbers to tries to gc
gcproc time.Duration // Accumulates canonical block processing for trie dumping
txLookupLimit uint64 // The maximum number of blocks from head whose tx indices are reserved
hc *HeaderChain
rmLogsFeed event.Feed
@ -181,7 +182,7 @@ type BlockChain struct {
// NewBlockChain returns a fully initialised block chain using information
// available in the database. It initialises the default Ethereum Validator and
// Processor.
func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(block *types.Block) bool) (*BlockChain, error) {
func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(block *types.Block) bool, txLookupLimit uint64) (*BlockChain, error) {
if cacheConfig == nil {
cacheConfig = &CacheConfig{
TrieCleanLimit: 256,
@ -200,6 +201,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
bc := &BlockChain{
chainConfig: chainConfig,
cacheConfig: cacheConfig,
txLookupLimit: txLookupLimit,
db: db,
triegc: prque.New(nil),
stateCache: state.NewDatabaseWithCache(db, cacheConfig.TrieCleanLimit),
@ -231,12 +233,6 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
// Initialize the chain with ancient data if it isn't empty.
if bc.empty() {
rawdb.InitBlockIndexFromFreezer(bc.db)
rawdb.WriteAncientTxLookupProgress(bc.db, 0) // Explicitly mark the missing of txlookup.
}
// Re-initialise all ancient txlookup indexes in the background.
if number := rawdb.ReadAncientTxLookupProgress(bc.db); number != nil {
// Genesis block doesn't have transaction, just ignore it.
go rawdb.InitTxsLookupFromFreezer(bc.db, *number+1)
}
if err := bc.loadLastState(); err != nil {
return nil, err
@ -294,6 +290,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
}
// Take ownership of this particular state
go bc.update()
go bc.updateTxIndices()
return bc, nil
}
@ -2041,6 +2038,101 @@ func (bc *BlockChain) update() {
}
}
// updateTxIndices is responsible for the construction and deletion of the
// transaction index.
//
// User can use flag `txlookuplimit` to specify a "recentness" block, below
// which ancient tx indices get deleted. If `txlookuplimit` is 0, it means
// all tx indices will be reserved.
//
// The user can adjust the txlookuplimit value for each launch, Geth will
// automatically construct the missing indices and delete the extra indices.
func (bc *BlockChain) updateTxIndices() {
var (
done chan struct{} // Non-nil if background unindexing or reindexing routine is active.
headCh = make(chan ChainHeadEvent)
)
sub := bc.SubscribeChainHeadEvent(headCh)
defer func() {
if sub != nil {
sub.Unsubscribe()
}
}()
// initialiseIndices inits txlookup indices into the database.
// If there already exists some indices, this function will find
// the oldest block which has been indexed and start indexing from
// this point.
initialiseIndices := func(head uint64) {
defer func() { done <- struct{}{} }()
from, to := uint64(0), head
if bc.txLookupLimit != 0 && head > bc.txLookupLimit {
from = head - bc.txLookupLimit
}
// Find oldest indexed block via binary search when we don't
// have this flag in database.
start := time.Now()
oldest := rawdb.FindOldestIndexedBlock(bc.db, from, to)
log.Debug("Find oldest indexed block", "oldest", oldest, "elapsed", common.PrettyDuration(time.Since(start)))
// Re-construct missing tx indices.
if oldest == nil {
rawdb.IndexTxLookup(bc.db, from, to) // No block has been indexed.
} else {
rawdb.IndexTxLookup(bc.db, from, *oldest)
}
// Delete useless tx indices if user requires.
if from > 0 {
rawdb.RemoveTxsLookup(bc.db, 0, from)
}
log.Debug("Initialised transaction indices", "elapsed", common.PrettyDuration(time.Since(start)))
}
// indexBlocks reindex or unindex transaction indices depends
// on user's requirement.
indexBlocks := func(oldest uint64, head uint64) {
defer func() { done <- struct{}{} }()
// All indices should be reserved.
if bc.txLookupLimit == 0 || head <= bc.txLookupLimit {
if oldest == 0 {
// Short circuit if nothing to delete.
return
} else {
// Reindex all indices if necessary
rawdb.IndexTxLookup(bc.db, 0, oldest)
return
}
}
if head-bc.txLookupLimit < oldest {
// Reindex a part of missing indices and rewind oldest indexed
// point to HEAD-limit
rawdb.IndexTxLookup(bc.db, head-bc.txLookupLimit, oldest)
} else {
// Unindex a part of stale indices and forward oldest indexed
// point to HEAD-limit
rawdb.RemoveTxsLookup(bc.db, oldest, head-bc.txLookupLimit)
}
}
for {
select {
case head := <-headCh:
if done == nil {
done = make(chan struct{})
if number := rawdb.ReadOldestIndexedBlock(bc.db); number == nil {
go initialiseIndices(head.Block.NumberU64())
} else {
go indexBlocks(*number, head.Block.NumberU64())
}
}
case <-done:
done = nil
case <-bc.quit:
return
}
}
}
// BadBlocks returns a list of the last 'bad blocks' that the client has seen on the network
func (bc *BlockChain) BadBlocks() []*types.Block {
blocks := make([]*types.Block, 0, bc.badBlocks.Len())

188
core/blockchain_test.go
View file

@ -54,7 +54,7 @@ func newCanonical(engine consensus.Engine, n int, full bool) (ethdb.Database, *B
)
// Initialize a fresh chain with only a genesis block
blockchain, _ := NewBlockChain(db, nil, params.AllEthashProtocolChanges, engine, vm.Config{}, nil)
blockchain, _ := NewBlockChain(db, nil, params.AllEthashProtocolChanges, engine, vm.Config{}, nil, 0)
// Create and inject the requested chain
if n == 0 {
return db, blockchain, nil
@ -509,7 +509,7 @@ func testReorgBadHashes(t *testing.T, full bool) {
blockchain.Stop()
// Create a new BlockChain and check that it rolled back the state.
ncm, err := NewBlockChain(blockchain.db, nil, blockchain.chainConfig, ethash.NewFaker(), vm.Config{}, nil)
ncm, err := NewBlockChain(blockchain.db, nil, blockchain.chainConfig, ethash.NewFaker(), vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create new chain manager: %v", err)
}
@ -621,7 +621,7 @@ func TestFastVsFullChains(t *testing.T) {
// Import the chain as an archive node for the comparison baseline
archiveDb := rawdb.NewMemoryDatabase()
gspec.MustCommit(archiveDb)
archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer archive.Stop()
if n, err := archive.InsertChain(blocks); err != nil {
@ -630,7 +630,7 @@ func TestFastVsFullChains(t *testing.T) {
// Fast import the chain as a non-archive node to test
fastDb := rawdb.NewMemoryDatabase()
gspec.MustCommit(fastDb)
fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer fast.Stop()
headers := make([]*types.Header, len(blocks))
@ -654,7 +654,7 @@ func TestFastVsFullChains(t *testing.T) {
t.Fatalf("failed to create temp freezer db: %v", err)
}
gspec.MustCommit(ancientDb)
ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer ancient.Stop()
if n, err := ancient.InsertHeaderChain(headers, 1); err != nil {
@ -750,7 +750,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) {
// Import the chain as an archive node and ensure all pointers are updated
archiveDb, delfn := makeDb()
defer delfn()
archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
if n, err := archive.InsertChain(blocks); err != nil {
t.Fatalf("failed to process block %d: %v", n, err)
}
@ -763,7 +763,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) {
// Import the chain as a non-archive node and ensure all pointers are updated
fastDb, delfn := makeDb()
defer delfn()
fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer fast.Stop()
headers := make([]*types.Header, len(blocks))
@ -783,7 +783,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) {
// Import the chain as a ancient-first node and ensure all pointers are updated
ancientDb, delfn := makeDb()
defer delfn()
ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer ancient.Stop()
if n, err := ancient.InsertHeaderChain(headers, 1); err != nil {
@ -802,7 +802,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) {
// Import the chain as a light node and ensure all pointers are updated
lightDb, delfn := makeDb()
defer delfn()
light, _ := NewBlockChain(lightDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
light, _ := NewBlockChain(lightDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
if n, err := light.InsertHeaderChain(headers, 1); err != nil {
t.Fatalf("failed to insert header %d: %v", n, err)
}
@ -871,7 +871,7 @@ func TestChainTxReorgs(t *testing.T) {
}
})
// Import the chain. This runs all block validation rules.
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
if i, err := blockchain.InsertChain(chain); err != nil {
t.Fatalf("failed to insert original chain[%d]: %v", i, err)
}
@ -941,7 +941,7 @@ func TestLogReorgs(t *testing.T) {
signer = types.NewEIP155Signer(gspec.Config.ChainID)
)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer blockchain.Stop()
rmLogsCh := make(chan RemovedLogsEvent)
@ -1018,7 +1018,7 @@ func TestLogRebirth(t *testing.T) {
}
}
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer blockchain.Stop()
logsCh := make(chan []*types.Log)
@ -1140,7 +1140,7 @@ func TestSideLogRebirth(t *testing.T) {
}
}
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer blockchain.Stop()
logsCh := make(chan []*types.Log)
@ -1195,7 +1195,7 @@ func TestReorgSideEvent(t *testing.T) {
signer = types.NewEIP155Signer(gspec.Config.ChainID)
)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer blockchain.Stop()
chain, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 3, func(i int, gen *BlockGen) {})
@ -1324,7 +1324,7 @@ func TestEIP155Transition(t *testing.T) {
genesis = gspec.MustCommit(db)
)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer blockchain.Stop()
blocks, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 4, func(i int, block *BlockGen) {
@ -1431,7 +1431,7 @@ func TestEIP161AccountRemoval(t *testing.T) {
}
genesis = gspec.MustCommit(db)
)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer blockchain.Stop()
blocks, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 3, func(i int, block *BlockGen) {
@ -1506,7 +1506,7 @@ func TestBlockchainHeaderchainReorgConsistency(t *testing.T) {
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@ -1550,7 +1550,7 @@ func TestTrieForkGC(t *testing.T) {
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@ -1589,7 +1589,7 @@ func TestLargeReorgTrieGC(t *testing.T) {
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@ -1649,7 +1649,7 @@ func TestBlockchainRecovery(t *testing.T) {
t.Fatalf("failed to create temp freezer db: %v", err)
}
gspec.MustCommit(ancientDb)
ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
headers := make([]*types.Header, len(blocks))
for i, block := range blocks {
@ -1668,7 +1668,7 @@ func TestBlockchainRecovery(t *testing.T) {
rawdb.WriteHeadFastBlockHash(ancientDb, midBlock.Hash())
// Reopen broken blockchain again
ancient, _ = NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
ancient, _ = NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer ancient.Stop()
if num := ancient.CurrentBlock().NumberU64(); num != 0 {
t.Errorf("head block mismatch: have #%v, want #%v", num, 0)
@ -1705,7 +1705,7 @@ func TestIncompleteAncientReceiptChainInsertion(t *testing.T) {
t.Fatalf("failed to create temp freezer db: %v", err)
}
gspec.MustCommit(ancientDb)
ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer ancient.Stop()
headers := make([]*types.Header, len(blocks))
@ -1762,7 +1762,7 @@ func TestLowDiffLongChain(t *testing.T) {
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@ -1809,7 +1809,7 @@ func testSideImport(t *testing.T, numCanonBlocksInSidechain, blocksBetweenCommon
blocks, _ := GenerateChain(params.TestChainConfig, genesis, engine, db, 2*TriesInMemory, nil)
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@ -1906,7 +1906,7 @@ func testInsertKnownChainData(t *testing.T, typ string) {
new(Genesis).MustCommit(chaindb)
defer os.RemoveAll(dir)
chain, err := NewBlockChain(chaindb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
chain, err := NewBlockChain(chaindb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@ -2029,7 +2029,7 @@ func getLongAndShortChains() (*BlockChain, []*types.Block, []*types.Block, error
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0)
if err != nil {
return nil, nil, nil, fmt.Errorf("failed to create tester chain: %v", err)
}
@ -2133,6 +2133,140 @@ func TestReorgToShorterRemovesCanonMappingHeaderChain(t *testing.T) {
}
}
func TestTransactionIndices(t *testing.T) {
// Configure and generate a sample block chain
var (
gendb = rawdb.NewMemoryDatabase()
key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
address = crypto.PubkeyToAddress(key.PublicKey)
funds = big.NewInt(1000000000)
gspec = &Genesis{Config: params.TestChainConfig, Alloc: GenesisAlloc{address: {Balance: funds}}}
genesis = gspec.MustCommit(gendb)
signer = types.NewEIP155Signer(gspec.Config.ChainID)
)
height := uint64(128)
blocks, receipts := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), gendb, int(height), func(i int, block *BlockGen) {
tx, err := types.SignTx(types.NewTransaction(block.TxNonce(address), common.Address{0x00}, big.NewInt(1000), params.TxGas, nil, nil), signer, key)
if err != nil {
panic(err)
}
block.AddTx(tx)
})
blocks2, _ := GenerateChain(gspec.Config, blocks[len(blocks)-1], ethash.NewFaker(), gendb, 10, nil)
check := func(oldest *uint64, chain *BlockChain) {
indexed := rawdb.ReadOldestIndexedBlock(chain.db)
if oldest == nil && indexed != nil {
t.Fatalf("Oldest indexded block mismatch, want nil, have %d", *indexed)
}
if oldest != nil && *indexed != *oldest {
t.Fatalf("Oldest indexded block mismatch, want %d, have %d", *oldest, *indexed)
}
if oldest != nil {
for i := *oldest; i <= chain.CurrentBlock().NumberU64(); i++ {
block := rawdb.ReadBlock(chain.db, rawdb.ReadCanonicalHash(chain.db, i), i)
if block.Transactions().Len() == 0 {
continue
}
for _, tx := range block.Transactions() {
if index := rawdb.ReadTxLookupEntry(chain.db, tx.Hash()); index == nil {
t.Fatalf("Miss transaction indice, number %d hash %s", i, tx.Hash().Hex())
}
}
}
for i := uint64(0); i < *oldest; i++ {
block := rawdb.ReadBlock(chain.db, rawdb.ReadCanonicalHash(chain.db, i), i)
if block.Transactions().Len() == 0 {
continue
}
for _, tx := range block.Transactions() {
if index := rawdb.ReadTxLookupEntry(chain.db, tx.Hash()); index != nil {
t.Fatalf("Transaction indice should be deleted, number %d hash %s", i, tx.Hash().Hex())
}
}
}
}
}
// Freezer style fast import the chain.
frdir, err := ioutil.TempDir("", "")
if err != nil {
t.Fatalf("failed to create temp freezer dir: %v", err)
}
defer os.Remove(frdir)
ancientDb, err := rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), frdir, "")
if err != nil {
t.Fatalf("failed to create temp freezer db: %v", err)
}
gspec.MustCommit(ancientDb)
// Import all blocks into ancient db
chain, err := NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
headers := make([]*types.Header, len(blocks))
for i, block := range blocks {
headers[i] = block.Header()
}
if n, err := chain.InsertHeaderChain(headers, 0); err != nil {
t.Fatalf("failed to insert header %d: %v", n, err)
}
if n, err := chain.InsertReceiptChain(blocks, receipts, 128); err != nil {
t.Fatalf("block %d: failed to insert into chain: %v", n, err)
}
chain.Stop()
ancientDb.Close()
// Reconstruct an ancient db with inserted ancient blocks.
ancientDb, err = rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), frdir, "")
if err != nil {
t.Fatalf("failed to create temp freezer db: %v", err)
}
gspec.MustCommit(ancientDb)
var oldest uint64
chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
chain.InsertChain(blocks2[:1]) // Feed chain a higher block to trigger indices updater.
time.Sleep(50 * time.Millisecond) // Wait for indices initialisation
check(&oldest, chain)
chain.Stop()
// Reconstruct a blockchain which only reserves HEAD-64 tx indices
chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 64)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
chain.InsertChain(blocks2[1:2]) // Feed chain a higher block to trigger indices updater.
time.Sleep(50 * time.Millisecond) // Wait for indices initialisation
oldest = chain.CurrentBlock().NumberU64() - 64
check(&oldest, chain)
chain.Stop()
// Reconstruct a block which only reserves HEAD-32 tx indices, shorten the indices history.
chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 32)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
chain.InsertChain(blocks2[2:3]) // Feed chain a higher block to trigger indices updater.
time.Sleep(50 * time.Millisecond) // Wait for indices initialisation
oldest = chain.CurrentBlock().NumberU64() - 32
check(&oldest, chain)
chain.Stop()
// Reconstruct a block which only reserves all tx indices, extends the indices history
chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
chain.InsertChain(blocks2[3:4]) // Feed chain a higher block to trigger indices updater.
time.Sleep(50 * time.Millisecond) // Wait for indices initialisation
oldest = 0
check(&oldest, chain)
}
// Benchmarks large blocks with value transfers to non-existing accounts
func benchmarkLargeNumberOfValueToNonexisting(b *testing.B, numTxs, numBlocks int, recipientFn func(uint64) common.Address, dataFn func(uint64) []byte) {
var (
@ -2178,7 +2312,7 @@ func benchmarkLargeNumberOfValueToNonexisting(b *testing.B, numTxs, numBlocks in
diskdb := rawdb.NewMemoryDatabase()
gspec.MustCommit(diskdb)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0)
if err != nil {
b.Fatalf("failed to create tester chain: %v", err)
}

2
core/chain_makers_test.go
View file

@ -79,7 +79,7 @@ func ExampleGenerateChain() {
})
// Import the chain. This runs all block validation rules.
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0)
defer blockchain.Stop()
if i, err := blockchain.InsertChain(chain); err != nil {

12
core/dao_test.go
View file

@ -45,7 +45,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
proConf.DAOForkBlock = forkBlock
proConf.DAOForkSupport = true
proBc, _ := NewBlockChain(proDb, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil)
proBc, _ := NewBlockChain(proDb, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil, 0)
defer proBc.Stop()
conDb := rawdb.NewMemoryDatabase()
@ -55,7 +55,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
conConf.DAOForkBlock = forkBlock
conConf.DAOForkSupport = false
conBc, _ := NewBlockChain(conDb, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil)
conBc, _ := NewBlockChain(conDb, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil, 0)
defer conBc.Stop()
if _, err := proBc.InsertChain(prefix); err != nil {
@ -69,7 +69,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
// Create a pro-fork block, and try to feed into the no-fork chain
db = rawdb.NewMemoryDatabase()
gspec.MustCommit(db)
bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil)
bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil, 0)
defer bc.Stop()
blocks := conBc.GetBlocksFromHash(conBc.CurrentBlock().Hash(), int(conBc.CurrentBlock().NumberU64()))
@ -94,7 +94,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
// Create a no-fork block, and try to feed into the pro-fork chain
db = rawdb.NewMemoryDatabase()
gspec.MustCommit(db)
bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil)
bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil, 0)
defer bc.Stop()
blocks = proBc.GetBlocksFromHash(proBc.CurrentBlock().Hash(), int(proBc.CurrentBlock().NumberU64()))
@ -120,7 +120,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
// Verify that contra-forkers accept pro-fork extra-datas after forking finishes
db = rawdb.NewMemoryDatabase()
gspec.MustCommit(db)
bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil)
bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil, 0)
defer bc.Stop()
blocks := conBc.GetBlocksFromHash(conBc.CurrentBlock().Hash(), int(conBc.CurrentBlock().NumberU64()))
@ -140,7 +140,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
// Verify that pro-forkers accept contra-fork extra-datas after forking finishes
db = rawdb.NewMemoryDatabase()
gspec.MustCommit(db)
bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil)
bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil, 0)
defer bc.Stop()
blocks = proBc.GetBlocksFromHash(proBc.CurrentBlock().Hash(), int(proBc.CurrentBlock().NumberU64()))

2
core/genesis_test.go
View file

@ -120,7 +120,7 @@ func TestSetupGenesis(t *testing.T) {
// Advance to block #4, past the homestead transition block of customg.
genesis := oldcustomg.MustCommit(db)
bc, _ := NewBlockChain(db, nil, oldcustomg.Config, ethash.NewFullFaker(), vm.Config{}, nil)
bc, _ := NewBlockChain(db, nil, oldcustomg.Config, ethash.NewFullFaker(), vm.Config{}, nil, 0)
defer bc.Stop()
blocks, _ := GenerateChain(oldcustomg.Config, genesis, ethash.NewFaker(), db, 4, nil)

63
core/rawdb/accessors_chain.go
View file

@ -171,10 +171,11 @@ func WriteFastTrieProgress(db ethdb.KeyValueWriter, count uint64) {
}
}
// ReadAncientTxLookupProgress retrieves the number of ancient blocks which
// txlookup has been inserted to allow reporting correct numbers across restarts.
func ReadAncientTxLookupProgress(db ethdb.KeyValueReader) *uint64 {
data, _ := db.Get(ancientTxLookupProgressKey)
// ReadOldestIndexedBlock retrieves the number of oldest indexed block
// whose transaction indices has been indexed. If the corresponding entry
// is non-existent in database it means the indexing has been finished.
func ReadOldestIndexedBlock(db ethdb.KeyValueReader) *uint64 {
data, _ := db.Get(oldestIndexedBlockKey)
if len(data) != 8 {
return nil
}
@ -182,18 +183,11 @@ func ReadAncientTxLookupProgress(db ethdb.KeyValueReader) *uint64 {
return &number
}
// WriteAncientTxLookupProgress stores the ancient txlookup process counter to support
// retrieving it across restarts.
func WriteAncientTxLookupProgress(db ethdb.KeyValueWriter, number uint64) {
if err := db.Put(ancientTxLookupProgressKey, encodeBlockNumber(number)); err != nil {
log.Crit("Failed to store head number of txlookup", "err", err)
}
}
// DeleteAncientTxLookupProgress deletes the ancient txlookup progress.
func DeleteAncientTxLookupProgress(db ethdb.KeyValueWriter) {
if err := db.Delete(ancientTxLookupProgressKey); err != nil {
log.Crit("Failed to delete ancient txlookup progress entry", "err", err)
// WriteOldestIndexedBlock stores the number of oldest indexed block
// into database.
func WriteOldestIndexedBlock(db ethdb.KeyValueWriter, number uint64) {
if err := db.Put(oldestIndexedBlockKey, encodeBlockNumber(number)); err != nil {
log.Crit("Failed to store the number of oldest indexed block", "err", err)
}
}
@ -584,3 +578,40 @@ func FindCommonAncestor(db ethdb.Reader, a, b *types.Header) *types.Header {
}
return a
}
// FindOldestIndexedBlock binary searches the oldest block which has been indexed.
// We will always ensures that if Bi is indexed, then Bi+1 must has been indexed.
//
// If no block has been indexed, then the returned value is to+1.
//
// The block doesn't contain any transaction will be regarded as unindexed. It can
// cause the blocks before this block can be reindexed.
func FindOldestIndexedBlock(db ethdb.Reader, from uint64, to uint64) *uint64 {
low, high := from, to+1
check := func(number uint64) bool {
block := ReadBlock(db, ReadCanonicalHash(db, number), number)
if block == nil {
log.Crit("Failed to retrieve block from database", "number", number)
}
if block.Transactions().Len() == 0 {
return false
}
if ReadTxLookupEntry(db, block.Transactions()[0].Hash()) == nil {
return false
}
return true
}
for low != high {
mid := (low + high) / 2
if !check(mid) {
low = mid + 1
} else {
high = mid
}
}
if low == to+1 {
return nil
}
return &low
}

49
core/rawdb/accessors_chain_test.go
View file

@ -358,3 +358,52 @@ func checkReceiptsRLP(have, want types.Receipts) error {
}
return nil
}
func TestFindOldestIndexedBlock(t *testing.T) {
var cases = []struct {
empty bool
oldest uint64
height uint64
nilBlocks map[uint64]bool
expect uint64
}{
{true, 0, 10, nil, 0}, // No block has been indexed
{false, 0, 10, nil, 1}, // Genesis block doesn't have indices
{false, 1, 10, nil, 1},
{false, 4, 10, nil, 4},
{false, 5, 10, nil, 5},
{false, 6, 10, nil, 6},
{false, 10, 10, nil, 10},
{false, 3, 10, map[uint64]bool{4: true, 6: true, 8: true}, 5},
}
for cid, c := range cases {
var (
db = NewMemoryDatabase()
block *types.Block
)
for i := uint64(0); i <= c.height; i++ {
if i == 0 {
block = types.NewBlock(&types.Header{Number: big.NewInt(int64(i))}, nil, nil, nil) // Empty genesis block
} else {
tx := types.NewTransaction(i, common.BytesToAddress([]byte{0x11}), big.NewInt(111), 1111, big.NewInt(11111), []byte{0x11, 0x11, 0x11})
txset := []*types.Transaction{tx}
if c.nilBlocks != nil && c.nilBlocks[i] {
txset = nil
}
block = types.NewBlock(&types.Header{Number: big.NewInt(int64(i))}, txset, nil, nil)
}
WriteBlock(db, block)
WriteCanonicalHash(db, block.Hash(), block.NumberU64())
if !c.empty && block.NumberU64() >= c.oldest {
WriteTxLookupEntries(db, block)
}
}
res := FindOldestIndexedBlock(db, 0, c.height)
if c.empty && res != nil {
t.Fatalf("Case %d failed, oldest block mismatch, want nil, have %d", cid, *res)
}
if !c.empty && *res != c.expect {
t.Fatalf("Case %d failed, oldest block mismatch, want %d, have %d", cid, c.expect, *res)
}
}
}

10
core/rawdb/accessors_indexes.go
View file

@ -63,6 +63,16 @@ func WriteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) {
}
}
// DeleteTxLookupEntries removes all transaction lookup indices contained in
// given block.
func DeleteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) {
for _, tx := range block.Transactions() {
if err := db.Delete(txLookupKey(tx.Hash())); err != nil {
log.Crit("Failed to delete transaction lookup entry", "err", err)
}
}
}
// DeleteTxLookupEntry removes all transaction data associated with a hash.
func DeleteTxLookupEntry(db ethdb.KeyValueWriter, hash common.Hash) {
db.Delete(txLookupKey(hash))

230
core/rawdb/chain_iterator.go Normal file
View file

@ -0,0 +1,230 @@
// Copyright 2019 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 rawdb
import (
"errors"
"math"
"runtime"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/prque"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
)
type (
prepareCallback func(*types.Block) // The callback for customized prepare operation.
actionCallback func(ethdb.Batch, *types.Block) // The callback for customized action.
)
// iterateCanonicalChain iterates the specified range canonical chain and then apply
// the given action callback.
// Note for forward iteration, the range is [from, to), otherwise the range is (from, to].
func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string, prepare prepareCallback, action actionCallback, reverse bool, report bool) error {
// Short circuit if the action is nil.
if action == nil {
return nil
}
// Short circuit if the iteration range is invalid.
if from >= to {
return nil
}
// Spawn multi-routines, iterate over the specified blocks and invoke prepare
// callback concurrently.
var (
number int64
results = make(chan *types.Block, 4*runtime.NumCPU())
)
if !reverse {
number = int64(from - 1)
} else {
number = int64(to + 1)
}
abort := make(chan struct{})
defer close(abort)
for i := 0; i < runtime.NumCPU(); i++ {
go func() {
for {
// Fetch the next task number, terminating if everything's done
var n int64
if !reverse {
n = atomic.AddInt64(&number, 1)
if n >= int64(to) {
return
}
} else {
n = atomic.AddInt64(&number, -1)
if n <= int64(from) {
return
}
}
block := ReadBlock(db, ReadCanonicalHash(db, uint64(n)), uint64(n))
if prepare != nil && block != nil {
prepare(block)
}
// Feed the block to the aggregator, or abort on interrupt
select {
case results <- block:
case <-abort:
return
}
}
}()
}
// Reassemble the blocks into a contiguous stream and apply the action callback.
var (
next, first, last int64
queue = prque.New(nil)
batch = db.NewBatch()
start = time.Now()
logged time.Time
)
if !reverse {
next, first, last = int64(from), int64(from), int64(to)
} else {
next, first, last = int64(to), int64(to), int64(from)
}
logFn := log.Debug
if report {
logFn = log.Info
}
for i := from; i < to; i++ {
// Retrieve the next result and bail if it's nil
block := <-results
if block == nil {
return errors.New("broken database")
}
// Push the block into the import queue and process contiguous ranges
priority := -int64(block.NumberU64())
if reverse {
priority = int64(block.NumberU64())
}
queue.Push(block, priority)
for !queue.Empty() {
// If the next available item is gapped, return
if _, priority := queue.Peek(); !reverse && -priority != next || reverse && priority != next {
break
}
// Next block available, pop it off and index it
block = queue.PopItem().(*types.Block)
if !reverse {
next++
} else {
next--
}
// Invoke action to inject specified data into key-value database.
action(batch, block)
// If enough data was accumulated in memory or we're at the last block, dump to disk
if batch.ValueSize() > ethdb.IdealBatchSize || next == last {
if err := batch.Write(); err != nil {
return err
}
batch.Reset()
}
// If we've spent too much time already, notify the user of what we're doing
if time.Since(logged) > 8*time.Second {
logFn("Iterating canonical chain", "type", typ, "reserve", reverse, "number", block.Number(), "hash", block.Hash(), "total", int64(math.Abs(float64(next-first))), "elapsed", common.PrettyDuration(time.Since(start)))
logged = time.Now()
}
}
}
logFn("Iterated canonical chain", "type", typ, "reverse", reverse, "total", to-from, "elapsed", common.PrettyDuration(time.Since(start)))
return nil
}
// InitBlockIndexFromFreezer reinitializes an empty database from a previous batch
// of frozen ancient blocks. The method iterates over all the frozen blocks and
// injects into the database the block hash->number mappings and the transaction
// lookup entries.
func InitBlockIndexFromFreezer(db ethdb.Database) {
// If we can't access the freezer or it's empty, abort
frozen, err := db.Ancients()
if err != nil || frozen == 0 {
return
}
// hashBlock calculates block hash in advance using the multi-routine's concurrent
// computing power.
hashBlock := func(block *types.Block) { block.Hash() }
// writeIndex injects hash <-> number mapping into the database.
writeIndex := func(batch ethdb.Batch, block *types.Block) { WriteHeaderNumber(batch, block.Hash(), block.NumberU64()) }
if err := iterateCanonicalChain(db, 0, frozen, "blocks", hashBlock, writeIndex, false, true); err != nil {
log.Crit("Failed to iterate canonical chain", "err", err)
}
hash := ReadCanonicalHash(db, frozen-1)
WriteHeadHeaderHash(db, hash)
WriteHeadFastBlockHash(db, hash)
log.Info("Initialized chain from ancient data", "number", frozen-1, "hash", hash)
}
// IndexTxLookup initializes txlookup indices of the specified range blocks into the database.
//
// This function iterates canonical chain in reverse order, it has two advantages:
// * If Geth crashes during the indexing without writing the oldest flag, we can
// binary search to quickly locate the oldest indexed block
// * We can write oldest indexed block flag periodically even without the whole
// indexing procedure is finished. So that we can resume indexing procedure next
// time quickly.
func IndexTxLookup(db ethdb.Database, from uint64, to uint64) {
// hashTxs calculates transaction hash in advance using the multi-routine's
// concurrent computing power.
hashTxs := func(block *types.Block) {
for _, tx := range block.Transactions() {
tx.Hash()
}
}
// writeIndices injects txlookup indices into the database.
writeIndices := func(batch ethdb.Batch, block *types.Block) {
WriteTxLookupEntries(batch, block)
if block.NumberU64()%1000000 == 0 {
WriteOldestIndexedBlock(batch, block.NumberU64())
}
}
if err := iterateCanonicalChain(db, from, to, "txlookup", hashTxs, writeIndices, true, true); err != nil {
log.Crit("Failed to iterate canonical chain", "err", err)
}
WriteOldestIndexedBlock(db, from)
log.Info("Constructed transaction indices", "from", from, "to", to, "count", to-from)
}
// RemoveTxsLookup removes txlookup indices of the specified range blocks.
func RemoveTxsLookup(db ethdb.Database, from uint64, to uint64) {
// Write flag first and then unindex the transaction indices. Some indices
// will be left in the database if crash happens but it's fine.
WriteOldestIndexedBlock(db, to)
if from+1 == to {
hash := ReadCanonicalHash(db, from)
DeleteTxLookupEntries(db, ReadBlock(db, hash, from))
log.Debug("Removed transaction indices", "number", from, "hash", hash)
} else {
deleteIndices := func(batch ethdb.Batch, block *types.Block) { DeleteTxLookupEntries(batch, block) }
if err := iterateCanonicalChain(db, from, to, "txlookup", nil, deleteIndices, false, false); err != nil {
log.Crit("Failed to iterate canonical chain", "err", err)
}
log.Debug("Removed transaction indices", "from", from, "to", to, "count", to-from)
}
}

180
core/rawdb/initer.go
View file

@ -1,180 +0,0 @@
// Copyright 2019 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 rawdb
import (
"errors"
"runtime"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/prque"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
)
type (
initPrepare func(*types.Block) // The callback for customized prepare operation.
initAction func(ethdb.Batch, *types.Block) // The callback for customized initialisation action.
)
// iterateAncient iterates the specified range blocks from ancient database
// and then apply initialisation action.
func iterateAncient(db ethdb.Database, from uint64, typ string, prepare initPrepare, action initAction) error {
// Short circuit if the init action is nil.
if action == nil {
return nil
}
// If we can't access the freezer or it's empty, abort
frozen, err := db.Ancients()
if err != nil || frozen == 0 {
return err
}
// Spawn multi-routines, iterate over the specified blocks and invoke prepare
// callback concurrently.
var (
number uint64
results = make(chan *types.Block, 4*runtime.NumCPU())
)
if from == 0 {
number = ^uint64(0) // -1
} else {
number = from - 1
}
abort := make(chan struct{})
defer close(abort)
for i := 0; i < runtime.NumCPU(); i++ {
go func() {
for {
// Fetch the next task number, terminating if everything's done
n := atomic.AddUint64(&number, 1)
if n >= frozen {
return
}
// Retrieve the block from the freezer (no need for the hash, we pull by
// number from the freezer).
block := ReadBlock(db, common.Hash{}, n)
if prepare != nil && block != nil {
prepare(block)
}
// Feed the block to the aggregator, or abort on interrupt
select {
case results <- block:
case <-abort:
return
}
}
}()
}
// Reassemble the blocks into a contiguous stream and apply the action callback.
var (
queue = prque.New(nil)
next = int64(from)
batch = db.NewBatch()
start = time.Now()
logged time.Time
)
for i := from; i < frozen; i++ {
// Retrieve the next result and bail if it's nil
block := <-results
if block == nil {
return errors.New("broken database")
}
// Push the block into the import queue and process contiguous ranges
queue.Push(block, -int64(block.NumberU64()))
for !queue.Empty() {
// If the next available item is gapped, return
if _, priority := queue.Peek(); -priority != next {
break
}
// Next block available, pop it off and index it
block = queue.PopItem().(*types.Block)
next++
// Invoke action to inject specified data into key-value database.
action(batch, block)
// If enough data was accumulated in memory or we're at the last block, dump to disk
if batch.ValueSize() > ethdb.IdealBatchSize || uint64(next) == frozen {
if err := batch.Write(); err != nil {
return err
}
batch.Reset()
}
// If we've spent too much time already, notify the user of what we're doing
if time.Since(logged) > 8*time.Second {
log.Info("Initializing chain from ancient data", "type", typ, "number", block.Number(), "hash", block.Hash(), "total", uint64(next)-from, "elapsed", common.PrettyDuration(time.Since(start)))
logged = time.Now()
}
}
}
log.Info("Initialized chain from ancient data", "type", typ, "number", frozen-from, "elapsed", common.PrettyDuration(time.Since(start)))
return nil
}
// InitBlockIndexFromFreezer reinitializes an empty database from a previous batch
// of frozen ancient blocks. The method iterates over all the frozen blocks and
// injects into the database the block hash->number mappings and the transaction
// lookup entries.
func InitBlockIndexFromFreezer(db ethdb.Database) error {
// If we can't access the freezer or it's empty, abort
frozen, err := db.Ancients()
if err != nil || frozen == 0 {
return err
}
// hashBlock calculates block hash in advance using the multi-routine's concurrent
// computing power.
hashBlock := func(block *types.Block) { block.Hash() }
// writeIndex injects hash <-> number mapping into the database.
writeIndex := func(batch ethdb.Batch, block *types.Block) { WriteHeaderNumber(batch, block.Hash(), block.NumberU64()) }
if err := iterateAncient(db, 0, "blocks", hashBlock, writeIndex); err != nil {
return err
}
hash := ReadCanonicalHash(db, frozen-1)
WriteHeadHeaderHash(db, hash)
WriteHeadFastBlockHash(db, hash)
return nil
}
// InitTxsLookupFromFreezer initializes txlookup indexes in the database.
func InitTxsLookupFromFreezer(db ethdb.Database, from uint64) error {
// hashTxs calculates transaction hash in advance using the multi-routine's
// concurrent computing power.
hashTxs := func(block *types.Block) {
for _, tx := range block.Transactions() {
tx.Hash()
}
}
// writeIndex injects txlookup indexes into the database.
writeIndex := func(batch ethdb.Batch, block *types.Block) {
WriteTxLookupEntries(batch, block)
if block.NumberU64()%10000 == 0 {
WriteAncientTxLookupProgress(batch, block.NumberU64())
}
}
if err := iterateAncient(db, from, "txlookup", hashTxs, writeIndex); err != nil {
return err
}
DeleteAncientTxLookupProgress(db) // Mark all txlookup indexes of ancient blocks have been inserted.
return nil
}

4
core/rawdb/schema.go
View file

@ -41,8 +41,8 @@ var (
// fastTrieProgressKey tracks the number of trie entries imported during fast sync.
fastTrieProgressKey = []byte("TrieSync")
// ancientTxLookupProgressKey tracks the progress of ancient txs lookup insertion.
ancientTxLookupProgressKey = []byte("AncientTxsLookup")
// oldestIndexedBlockKey tracks the oldest block whose transaction indices(txlookup) has been indexed.
oldestIndexedBlockKey = []byte("OldestIndexedBlock")
// Data item prefixes (use single byte to avoid mixing data types, avoid `i`, used for indexes).
headerPrefix = []byte("h") // headerPrefix + num (uint64 big endian) + hash -> header

2
eth/backend.go
View file

@ -186,7 +186,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
TrieTimeLimit: config.TrieTimeout,
}
)
eth.blockchain, err = core.NewBlockChain(chainDb, cacheConfig, chainConfig, eth.engine, vmConfig, eth.shouldPreserve)
eth.blockchain, err = core.NewBlockChain(chainDb, cacheConfig, chainConfig, eth.engine, vmConfig, eth.shouldPreserve, config.TxLookupLimit)
if err != nil {
return nil, err
}

2
eth/config.go
View file

@ -98,6 +98,8 @@ type Config struct {
NoPruning bool // Whether to disable pruning and flush everything to disk
NoPrefetch bool // Whether to disable prefetching and only load state on demand
TxLookupLimit uint64 `toml:",omitempty"` // The maximum number of blocks from head whose tx indices are reserved.
// Whitelist of required block number -> hash values to accept
Whitelist map[uint64]common.Hash `toml:"-"`

6
eth/gen_config.go
View file

@ -23,6 +23,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
SyncMode downloader.SyncMode
NoPruning bool
NoPrefetch bool
TxLookupLimit uint64 `toml:",omitempty"`
Whitelist map[uint64]common.Hash `toml:"-"`
LightServ int `toml:",omitempty"`
LightIngress int `toml:",omitempty"`
@ -56,6 +57,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
enc.SyncMode = c.SyncMode
enc.NoPruning = c.NoPruning
enc.NoPrefetch = c.NoPrefetch
enc.TxLookupLimit = c.TxLookupLimit
enc.Whitelist = c.Whitelist
enc.LightServ = c.LightServ
enc.LightIngress = c.LightIngress
@ -93,6 +95,7 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
SyncMode *downloader.SyncMode
NoPruning *bool
NoPrefetch *bool
TxLookupLimit *uint64 `toml:",omitempty"`
Whitelist map[uint64]common.Hash `toml:"-"`
LightServ *int `toml:",omitempty"`
LightIngress *int `toml:",omitempty"`
@ -139,6 +142,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
if dec.NoPrefetch != nil {
c.NoPrefetch = *dec.NoPrefetch
}
if dec.TxLookupLimit != nil {
c.TxLookupLimit = *dec.TxLookupLimit
}
if dec.Whitelist != nil {
c.Whitelist = dec.Whitelist
}

4
eth/handler_test.go
View file

@ -489,7 +489,7 @@ func testCheckpointChallenge(t *testing.T, syncmode downloader.SyncMode, checkpo
}
}
// Create a checkpoint aware protocol manager
blockchain, err := core.NewBlockChain(db, nil, config, ethash.NewFaker(), vm.Config{}, nil)
blockchain, err := core.NewBlockChain(db, nil, config, ethash.NewFaker(), vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create new blockchain: %v", err)
}
@ -576,7 +576,7 @@ func testBroadcastBlock(t *testing.T, totalPeers, broadcastExpected int) {
gspec = &core.Genesis{Config: config}
genesis = gspec.MustCommit(db)
)
blockchain, err := core.NewBlockChain(db, nil, config, pow, vm.Config{}, nil)
blockchain, err := core.NewBlockChain(db, nil, config, pow, vm.Config{}, nil, 0)
if err != nil {
t.Fatalf("failed to create new blockchain: %v", err)
}

2
eth/helper_test.go
View file

@ -60,7 +60,7 @@ func newTestProtocolManager(mode downloader.SyncMode, blocks int, generator func
Alloc: core.GenesisAlloc{testBank: {Balance: big.NewInt(1000000)}},
}
genesis = gspec.MustCommit(db)
blockchain, _ = core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}, nil)
blockchain, _ = core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}, nil, 0)
)
chain, _ := core.GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, blocks, generator)
if _, err := blockchain.InsertChain(chain); err != nil {

2
light/odr_test.go
View file

@ -257,7 +257,7 @@ func testChainOdr(t *testing.T, protocol int, fn odrTestFn) {
)
gspec.MustCommit(ldb)
// Assemble the test environment
blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil)
blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil, 0)
gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), sdb, 4, testChainGen)
if _, err := blockchain.InsertChain(gchain); err != nil {
t.Fatal(err)

2
light/trie_test.go
View file

@ -40,7 +40,7 @@ func TestNodeIterator(t *testing.T) {
genesis = gspec.MustCommit(fulldb)
)
gspec.MustCommit(lightdb)
blockchain, _ := core.NewBlockChain(fulldb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil)
blockchain, _ := core.NewBlockChain(fulldb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil, 0)
gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), fulldb, 4, testChainGen)
if _, err := blockchain.InsertChain(gchain); err != nil {
panic(err)

2
light/txpool_test.go
View file

@ -88,7 +88,7 @@ func TestTxPool(t *testing.T) {
)
gspec.MustCommit(ldb)
// Assemble the test environment
blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil)
blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil, 0)
gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), sdb, poolTestBlocks, txPoolTestChainGen)
if _, err := blockchain.InsertChain(gchain); err != nil {
panic(err)

2
miner/worker_test.go
View file

@ -103,7 +103,7 @@ func newTestWorkerBackend(t *testing.T, chainConfig *params.ChainConfig, engine
}
genesis := gspec.MustCommit(db)
chain, _ := core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}, nil)
chain, _ := core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}, nil, 0)
txpool := core.NewTxPool(testTxPoolConfig, chainConfig, chain)
// Generate a small n-block chain and an uncle block for it

2
tests/block_test_util.go
View file

@ -118,7 +118,7 @@ func (t *BlockTest) Run() error {
} else {
engine = ethash.NewShared()
}
chain, err := core.NewBlockChain(db, &core.CacheConfig{TrieCleanLimit: 0}, config, engine, vm.Config{}, nil)
chain, err := core.NewBlockChain(db, &core.CacheConfig{TrieCleanLimit: 0}, config, engine, vm.Config{}, nil, 0)
if err != nil {
return err
}