diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 2536d42b31..a321891d3f 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -230,12 +230,18 @@ func initGenesis(ctx *cli.Context) error { defer chaindb.Close() // if the trie data dir has been set, new trie db with a new state database - if ctx.IsSet(utils.SeparateDBFlag.Name) { + if ctx.IsSet(utils.MultiDataBaseFlag.Name) { statediskdb, dbErr := stack.OpenDatabaseWithFreezer(name+"/state", 0, 0, "", "", false) if dbErr != nil { utils.Fatalf("Failed to open separate trie database: %v", dbErr) } chaindb.SetStateStore(statediskdb) + blockdb, err := stack.OpenDatabaseWithFreezer(name+"/block", 0, 0, "", "", false) + if err != nil { + utils.Fatalf("Failed to open separate block database: %v", err) + } + chaindb.SetBlockStore(blockdb) + log.Warn("Multi-database is an experimental feature") } triedb := utils.MakeTrieDatabase(ctx, chaindb, ctx.Bool(utils.CachePreimagesFlag.Name), false, genesis.IsVerkle()) @@ -276,6 +282,13 @@ func dumpGenesis(ctx *cli.Context) error { } continue } + // set the separate state & block database + if stack.CheckIfMultiDataBase() && err == nil { + stateDiskDb := utils.MakeStateDataBase(ctx, stack, true, false) + db.SetStateStore(stateDiskDb) + blockDb := utils.MakeBlockDatabase(ctx, stack, true, false) + db.SetBlockStore(blockDb) + } genesis, err := core.ReadGenesis(db) if err != nil { utils.Fatalf("failed to read genesis: %s", err) @@ -534,7 +547,7 @@ func parseDumpConfig(ctx *cli.Context, stack *node.Node, db ethdb.Database) (*st arg := ctx.Args().First() if hashish(arg) { hash := common.HexToHash(arg) - if number := rawdb.ReadHeaderNumber(db, hash); number != nil { + if number := rawdb.ReadHeaderNumber(db.BlockStore(), hash); number != nil { header = rawdb.ReadHeader(db, hash, *number) } else { return nil, common.Hash{}, fmt.Errorf("block %x not found", hash) @@ -593,6 +606,7 @@ func dump(ctx *cli.Context) error { if err != nil { return err } + defer db.Close() triedb := utils.MakeTrieDatabase(ctx, db, true, true, false) // always enable preimage lookup defer triedb.Close() diff --git a/cmd/geth/config.go b/cmd/geth/config.go index c82e062ae4..522e5e22f2 100644 --- a/cmd/geth/config.go +++ b/cmd/geth/config.go @@ -179,9 +179,6 @@ func makeFullNode(ctx *cli.Context) *node.Node { cfg.Eth.OverrideVerkle = &v } - if ctx.IsSet(utils.SeparateDBFlag.Name) && !stack.IsSeparatedDB() { - utils.Fatalf("Failed to locate separate database subdirectory when separatedb parameter has been set") - } backend, eth := utils.RegisterEthService(stack, &cfg.Eth) // Create gauge with geth system and build information diff --git a/cmd/geth/dbcmd.go b/cmd/geth/dbcmd.go index 79db88b0d7..d3f8018a83 100644 --- a/cmd/geth/dbcmd.go +++ b/cmd/geth/dbcmd.go @@ -374,7 +374,7 @@ func checkStateContent(ctx *cli.Context) error { db := utils.MakeChainDatabase(ctx, stack, true) defer db.Close() var ( - it = rawdb.NewKeyLengthIterator(db.NewIterator(prefix, start), 32) + it ethdb.Iterator hasher = crypto.NewKeccakState() got = make([]byte, 32) errs int @@ -382,6 +382,11 @@ func checkStateContent(ctx *cli.Context) error { startTime = time.Now() lastLog = time.Now() ) + if stack.CheckIfMultiDataBase() { + it = rawdb.NewKeyLengthIterator(db.StateStore().NewIterator(prefix, start), 32) + } else { + it = rawdb.NewKeyLengthIterator(db.NewIterator(prefix, start), 32) + } for it.Next() { count++ k := it.Key() @@ -428,9 +433,11 @@ func dbStats(ctx *cli.Context) error { defer db.Close() showLeveldbStats(db) - if db.StateStore() != nil { + if stack.CheckIfMultiDataBase() { fmt.Println("show stats of state store") showLeveldbStats(db.StateStore()) + fmt.Println("show stats of block store") + showLeveldbStats(db.BlockStore()) } return nil @@ -446,10 +453,11 @@ func dbCompact(ctx *cli.Context) error { log.Info("Stats before compaction") showLeveldbStats(db) - statediskdb := db.StateStore() - if statediskdb != nil { + if stack.CheckIfMultiDataBase() { fmt.Println("show stats of state store") - showLeveldbStats(statediskdb) + showLeveldbStats(db.StateStore()) + fmt.Println("show stats of block store") + showLeveldbStats(db.BlockStore()) } log.Info("Triggering compaction") @@ -458,8 +466,12 @@ func dbCompact(ctx *cli.Context) error { return err } - if statediskdb != nil { - if err := statediskdb.Compact(nil, nil); err != nil { + if stack.CheckIfMultiDataBase() { + if err := db.StateStore().Compact(nil, nil); err != nil { + log.Error("Compact err", "error", err) + return err + } + if err := db.BlockStore().Compact(nil, nil); err != nil { log.Error("Compact err", "error", err) return err } @@ -467,9 +479,11 @@ func dbCompact(ctx *cli.Context) error { log.Info("Stats after compaction") showLeveldbStats(db) - if statediskdb != nil { + if stack.CheckIfMultiDataBase() { fmt.Println("show stats of state store after compaction") - showLeveldbStats(statediskdb) + showLeveldbStats(db.StateStore()) + fmt.Println("show stats of block store after compaction") + showLeveldbStats(db.BlockStore()) } return nil } @@ -490,18 +504,18 @@ func dbGet(ctx *cli.Context) error { log.Info("Could not decode the key", "error", err) return err } - - statediskdb := db.StateStore() - data, err := db.Get(key) - if err != nil { - // if separate trie db exist, try to get it from separate db - if statediskdb != nil { - statedata, dberr := statediskdb.Get(key) - if dberr == nil { - fmt.Printf("key %#x: %#x\n", key, statedata) - return nil - } + opDb := db + if stack.CheckIfMultiDataBase() { + keyType := rawdb.DataTypeByKey(key) + if keyType == rawdb.StateDataType { + opDb = db.StateStore() + } else if keyType == rawdb.BlockDataType { + opDb = db.BlockStore() } + } + + data, err := opDb.Get(key) + if err != nil { log.Info("Get operation failed", "key", fmt.Sprintf("%#x", key), "error", err) return err } @@ -525,11 +539,21 @@ func dbDelete(ctx *cli.Context) error { log.Info("Could not decode the key", "error", err) return err } - data, err := db.Get(key) + opDb := db + if stack.CheckIfMultiDataBase() { + keyType := rawdb.DataTypeByKey(key) + if keyType == rawdb.StateDataType { + opDb = db.StateStore() + } else if keyType == rawdb.BlockDataType { + opDb = db.BlockStore() + } + } + + data, err := opDb.Get(key) if err == nil { fmt.Printf("Previous value: %#x\n", data) } - if err = db.Delete(key); err != nil { + if err = opDb.Delete(key); err != nil { log.Info("Delete operation returned an error", "key", fmt.Sprintf("%#x", key), "error", err) return err } @@ -563,11 +587,22 @@ func dbPut(ctx *cli.Context) error { log.Info("Could not decode the value", "error", err) return err } - data, err = db.Get(key) + + opDb := db + if stack.CheckIfMultiDataBase() { + keyType := rawdb.DataTypeByKey(key) + if keyType == rawdb.StateDataType { + opDb = db.StateStore() + } else if keyType == rawdb.BlockDataType { + opDb = db.BlockStore() + } + } + + data, err = opDb.Get(key) if err == nil { fmt.Printf("Previous value: %#x\n", data) } - return db.Put(key, value) + return opDb.Put(key, value) } // dbDumpTrie shows the key-value slots of a given storage trie @@ -580,7 +615,6 @@ func dbDumpTrie(ctx *cli.Context) error { db := utils.MakeChainDatabase(ctx, stack, true) defer db.Close() - triedb := utils.MakeTrieDatabase(ctx, db, false, true, false) defer triedb.Close() @@ -659,7 +693,7 @@ func freezerInspect(ctx *cli.Context) error { stack, _ := makeConfigNode(ctx) ancient := stack.ResolveAncient("chaindata", ctx.String(utils.AncientFlag.Name)) stack.Close() - return rawdb.InspectFreezerTable(ancient, freezer, table, start, end) + return rawdb.InspectFreezerTable(ancient, freezer, table, start, end, stack.CheckIfMultiDataBase()) } func importLDBdata(ctx *cli.Context) error { @@ -799,12 +833,11 @@ func showMetaData(ctx *cli.Context) error { defer stack.Close() db := utils.MakeChainDatabase(ctx, stack, true) defer db.Close() - - ancients, err := db.Ancients() + ancients, err := db.BlockStore().Ancients() if err != nil { fmt.Fprintf(os.Stderr, "Error accessing ancients: %v", err) } - data := rawdb.ReadChainMetadata(db) + data := rawdb.ReadChainMetadataFromMultiDatabase(db) data = append(data, []string{"frozen", fmt.Sprintf("%d items", ancients)}) data = append(data, []string{"snapshotGenerator", snapshot.ParseGeneratorStatus(rawdb.ReadSnapshotGenerator(db))}) if b := rawdb.ReadHeadBlock(db); b != nil { diff --git a/cmd/geth/snapshot.go b/cmd/geth/snapshot.go index 08a56dc00c..d7b37c6243 100644 --- a/cmd/geth/snapshot.go +++ b/cmd/geth/snapshot.go @@ -545,6 +545,7 @@ func dumpState(ctx *cli.Context) error { if err != nil { return err } + defer db.Close() triedb := utils.MakeTrieDatabase(ctx, db, false, true, false) defer triedb.Close() diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index ff4c4e226b..819796c4c8 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -94,10 +94,10 @@ var ( Value: flags.DirectoryString(node.DefaultDataDir()), Category: flags.EthCategory, } - SeparateDBFlag = &cli.BoolFlag{ - Name: "separatedb", - Usage: "Enable a separated trie database, it will be created within a subdirectory called state, " + - "Users can copy this state directory to another directory or disk, and then create a symbolic link to the state directory under the chaindata", + MultiDataBaseFlag = &cli.BoolFlag{ + Name: "multidatabase", + Usage: "Enable a separated state and block database, it will be created within two subdirectory called state and block, " + + "Users can copy this state or block directory to another directory or disk, and then create a symbolic link to the state directory under the chaindata", Category: flags.EthCategory, } RemoteDBFlag = &cli.StringFlag{ @@ -980,7 +980,7 @@ var ( DBEngineFlag, StateSchemeFlag, HttpHeaderFlag, - SeparateDBFlag, + MultiDataBaseFlag, } ) @@ -2078,9 +2078,11 @@ func MakeChainDatabase(ctx *cli.Context, stack *node.Node, readonly bool) ethdb. default: chainDb, err = stack.OpenDatabaseWithFreezer("chaindata", cache, handles, ctx.String(AncientFlag.Name), "", readonly) // set the separate state database - if stack.IsSeparatedDB() && err == nil { + if stack.CheckIfMultiDataBase() && err == nil { stateDiskDb := MakeStateDataBase(ctx, stack, readonly, false) chainDb.SetStateStore(stateDiskDb) + blockDb := MakeBlockDatabase(ctx, stack, readonly, false) + chainDb.SetBlockStore(blockDb) } } if err != nil { @@ -2092,7 +2094,7 @@ func MakeChainDatabase(ctx *cli.Context, stack *node.Node, readonly bool) ethdb. // MakeStateDataBase open a separate state database using the flags passed to the client and will hard crash if it fails. func MakeStateDataBase(ctx *cli.Context, stack *node.Node, readonly, disableFreeze bool) ethdb.Database { cache := ctx.Int(CacheFlag.Name) * ctx.Int(CacheDatabaseFlag.Name) / 100 - handles := MakeDatabaseHandles(ctx.Int(FDLimitFlag.Name)) / 2 + handles := MakeDatabaseHandles(ctx.Int(FDLimitFlag.Name)) * 90 / 100 statediskdb, err := stack.OpenDatabaseWithFreezer("chaindata/state", cache, handles, "", "", readonly) if err != nil { Fatalf("Failed to open separate trie database: %v", err) @@ -2100,6 +2102,17 @@ func MakeStateDataBase(ctx *cli.Context, stack *node.Node, readonly, disableFree return statediskdb } +// MakeBlockDatabase open a separate block database using the flags passed to the client and will hard crash if it fails. +func MakeBlockDatabase(ctx *cli.Context, stack *node.Node, readonly, disableFreeze bool) ethdb.Database { + cache := ctx.Int(CacheFlag.Name) * ctx.Int(CacheDatabaseFlag.Name) / 100 + handles := MakeDatabaseHandles(ctx.Int(FDLimitFlag.Name)) / 10 + blockDb, err := stack.OpenDatabaseWithFreezer("chaindata/block", cache, handles, "", "", readonly) + if err != nil { + Fatalf("Failed to open separate block database: %v", err) + } + return blockDb +} + // tryMakeReadOnlyDatabase try to open the chain database in read-only mode, // or fallback to write mode if the database is not initialized. func tryMakeReadOnlyDatabase(ctx *cli.Context, stack *node.Node) ethdb.Database { diff --git a/core/blockchain.go b/core/blockchain.go index 7c8ab3abc4..3328dc03db 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -472,7 +472,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis // into node seamlessly. func (bc *BlockChain) empty() bool { genesis := bc.genesisBlock.Hash() - for _, hash := range []common.Hash{rawdb.ReadHeadBlockHash(bc.db), rawdb.ReadHeadHeaderHash(bc.db), rawdb.ReadHeadFastBlockHash(bc.db)} { + for _, hash := range []common.Hash{rawdb.ReadHeadBlockHash(bc.db.BlockStore()), rawdb.ReadHeadHeaderHash(bc.db.BlockStore()), rawdb.ReadHeadFastBlockHash(bc.db)} { if hash != genesis { return false } @@ -484,7 +484,7 @@ func (bc *BlockChain) empty() bool { // assumes that the chain manager mutex is held. func (bc *BlockChain) loadLastState() error { // Restore the last known head block - head := rawdb.ReadHeadBlockHash(bc.db) + head := rawdb.ReadHeadBlockHash(bc.db.BlockStore()) if head == (common.Hash{}) { // Corrupt or empty database, init from scratch log.Warn("Empty database, resetting chain") @@ -503,7 +503,7 @@ func (bc *BlockChain) loadLastState() error { // Restore the last known head header headHeader := headBlock.Header() - if head := rawdb.ReadHeadHeaderHash(bc.db); head != (common.Hash{}) { + if head := rawdb.ReadHeadHeaderHash(bc.db.BlockStore()); head != (common.Hash{}) { if header := bc.GetHeaderByHash(head); header != nil { headHeader = header } @@ -605,10 +605,10 @@ func (bc *BlockChain) SetHeadWithTimestamp(timestamp uint64) error { func (bc *BlockChain) SetFinalized(header *types.Header) { bc.currentFinalBlock.Store(header) if header != nil { - rawdb.WriteFinalizedBlockHash(bc.db, header.Hash()) + rawdb.WriteFinalizedBlockHash(bc.db.BlockStore(), header.Hash()) headFinalizedBlockGauge.Update(int64(header.Number.Uint64())) } else { - rawdb.WriteFinalizedBlockHash(bc.db, common.Hash{}) + rawdb.WriteFinalizedBlockHash(bc.db.BlockStore(), common.Hash{}) headFinalizedBlockGauge.Update(0) } } @@ -983,10 +983,10 @@ func (bc *BlockChain) ResetWithGenesisBlock(genesis *types.Block) error { defer bc.chainmu.Unlock() // Prepare the genesis block and reinitialise the chain - batch := bc.db.NewBatch() - rawdb.WriteTd(batch, genesis.Hash(), genesis.NumberU64(), genesis.Difficulty()) - rawdb.WriteBlock(batch, genesis) - if err := batch.Write(); err != nil { + blockBatch := bc.db.BlockStore().NewBatch() + rawdb.WriteTd(blockBatch, genesis.Hash(), genesis.NumberU64(), genesis.Difficulty()) + rawdb.WriteBlock(blockBatch, genesis) + if err := blockBatch.Write(); err != nil { log.Crit("Failed to write genesis block", "err", err) } bc.writeHeadBlock(genesis) @@ -1047,12 +1047,13 @@ func (bc *BlockChain) ExportN(w io.Writer, first uint64, last uint64) error { // Note, this function assumes that the `mu` mutex is held! func (bc *BlockChain) writeHeadBlock(block *types.Block) { // Add the block to the canonical chain number scheme and mark as the head + rawdb.WriteCanonicalHash(bc.db.BlockStore(), block.Hash(), block.NumberU64()) + rawdb.WriteHeadHeaderHash(bc.db.BlockStore(), block.Hash()) + rawdb.WriteHeadBlockHash(bc.db.BlockStore(), block.Hash()) + batch := bc.db.NewBatch() - rawdb.WriteHeadHeaderHash(batch, block.Hash()) rawdb.WriteHeadFastBlockHash(batch, block.Hash()) - rawdb.WriteCanonicalHash(batch, block.Hash(), block.NumberU64()) rawdb.WriteTxLookupEntriesByBlock(batch, block) - rawdb.WriteHeadBlockHash(batch, block.Hash()) // Flush the whole batch into the disk, exit the node if failed if err := batch.Write(); err != nil { @@ -1307,24 +1308,24 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ // Delete block data from the main database. var ( - batch = bc.db.NewBatch() canonHashes = make(map[common.Hash]struct{}, len(blockChain)) + blockBatch = bc.db.BlockStore().NewBatch() ) for _, block := range blockChain { canonHashes[block.Hash()] = struct{}{} if block.NumberU64() == 0 { continue } - rawdb.DeleteCanonicalHash(batch, block.NumberU64()) - rawdb.DeleteBlockWithoutNumber(batch, block.Hash(), block.NumberU64()) + rawdb.DeleteCanonicalHash(blockBatch, block.NumberU64()) + rawdb.DeleteBlockWithoutNumber(blockBatch, block.Hash(), block.NumberU64()) } // Delete side chain hash-to-number mappings. for _, nh := range rawdb.ReadAllHashesInRange(bc.db, first.NumberU64(), last.NumberU64()) { if _, canon := canonHashes[nh.Hash]; !canon { - rawdb.DeleteHeader(batch, nh.Hash, nh.Number) + rawdb.DeleteHeader(blockBatch, nh.Hash, nh.Number) } } - if err := batch.Write(); err != nil { + if err := blockBatch.Write(); err != nil { return 0, err } stats.processed += int32(len(blockChain)) @@ -1336,6 +1337,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ var ( skipPresenceCheck = false batch = bc.db.NewBatch() + blockBatch = bc.db.BlockStore().NewBatch() ) for i, block := range blockChain { // Short circuit insertion if shutting down or processing failed @@ -1359,8 +1361,8 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ } } // Write all the data out into the database - rawdb.WriteBody(batch, block.Hash(), block.NumberU64(), block.Body()) - rawdb.WriteReceipts(batch, block.Hash(), block.NumberU64(), receiptChain[i]) + rawdb.WriteBody(blockBatch, block.Hash(), block.NumberU64(), block.Body()) + rawdb.WriteReceipts(blockBatch, block.Hash(), block.NumberU64(), receiptChain[i]) // Write everything belongs to the blocks into the database. So that // we can ensure all components of body is completed(body, receipts) @@ -1372,6 +1374,13 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ size += int64(batch.ValueSize()) batch.Reset() } + if blockBatch.ValueSize() >= ethdb.IdealBatchSize { + if err := blockBatch.Write(); err != nil { + return 0, err + } + size += int64(blockBatch.ValueSize()) + blockBatch.Reset() + } stats.processed++ } // Write everything belongs to the blocks into the database. So that @@ -1383,6 +1392,12 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ return 0, err } } + if blockBatch.ValueSize() > 0 { + size += int64(blockBatch.ValueSize()) + if err := blockBatch.Write(); err != nil { + return 0, err + } + } updateHead(blockChain[len(blockChain)-1]) return 0, nil } @@ -1427,10 +1442,10 @@ func (bc *BlockChain) writeBlockWithoutState(block *types.Block, td *big.Int) (e if bc.insertStopped() { return errInsertionInterrupted } - batch := bc.db.NewBatch() - rawdb.WriteTd(batch, block.Hash(), block.NumberU64(), td) - rawdb.WriteBlock(batch, block) - if err := batch.Write(); err != nil { + blockBatch := bc.db.BlockStore().NewBatch() + rawdb.WriteTd(blockBatch, block.Hash(), block.NumberU64(), td) + rawdb.WriteBlock(blockBatch, block) + if err := blockBatch.Write(); err != nil { log.Crit("Failed to write block into disk", "err", err) } return nil @@ -1464,7 +1479,7 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types. // // Note all the components of block(td, hash->number map, header, body, receipts) // should be written atomically. BlockBatch is used for containing all components. - blockBatch := bc.db.NewBatch() + blockBatch := bc.db.BlockStore().NewBatch() rawdb.WriteTd(blockBatch, block.Hash(), block.NumberU64(), externTd) rawdb.WriteBlock(blockBatch, block) rawdb.WriteReceipts(blockBatch, block.Hash(), block.NumberU64(), receipts) @@ -1769,7 +1784,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error) // state, but if it's this special case here(skip reexecution) we will lose // the empty receipt entry. if len(block.Transactions()) == 0 { - rawdb.WriteReceipts(bc.db, block.Hash(), block.NumberU64(), nil) + rawdb.WriteReceipts(bc.db.BlockStore(), block.Hash(), block.NumberU64(), nil) } else { log.Error("Please file an issue, skip known block execution without receipt", "hash", block.Hash(), "number", block.NumberU64()) @@ -2296,6 +2311,7 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Block) error { var ( indexesBatch = bc.db.NewBatch() diffs = types.HashDifference(deletedTxs, addedTxs) + blockBatch = bc.db.BlockStore().NewBatch() ) for _, tx := range diffs { rawdb.DeleteTxLookupEntry(indexesBatch, tx) @@ -2312,11 +2328,14 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Block) error { if hash == (common.Hash{}) { break } - rawdb.DeleteCanonicalHash(indexesBatch, i) + rawdb.DeleteCanonicalHash(blockBatch, i) } if err := indexesBatch.Write(); err != nil { log.Crit("Failed to delete useless indexes", "err", err) } + if err := blockBatch.Write(); err != nil { + log.Crit("Failed to delete useless indexes use block batch", "err", err) + } // Reset the tx lookup cache to clear stale txlookup cache. bc.txLookupCache.Purge() diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index 8a85800dd8..44aeeffbe5 100644 --- a/core/blockchain_reader.go +++ b/core/blockchain_reader.go @@ -214,7 +214,7 @@ func (bc *BlockChain) GetReceiptsByHash(hash common.Hash) types.Receipts { if receipts, ok := bc.receiptsCache.Get(hash); ok { return receipts } - number := rawdb.ReadHeaderNumber(bc.db, hash) + number := rawdb.ReadHeaderNumber(bc.db.BlockStore(), hash) if number == nil { return nil } diff --git a/core/chain_indexer.go b/core/chain_indexer.go index f5fce72588..6ccca69e52 100644 --- a/core/chain_indexer.go +++ b/core/chain_indexer.go @@ -227,8 +227,8 @@ func (c *ChainIndexer) eventLoop(currentHeader *types.Header, events chan ChainH // Reorg to the common ancestor if needed (might not exist in light sync mode, skip reorg then) // TODO(karalabe, zsfelfoldi): This seems a bit brittle, can we detect this case explicitly? - if rawdb.ReadCanonicalHash(c.chainDb, prevHeader.Number.Uint64()) != prevHash { - if h := rawdb.FindCommonAncestor(c.chainDb, prevHeader, header); h != nil { + if rawdb.ReadCanonicalHash(c.chainDb.BlockStore(), prevHeader.Number.Uint64()) != prevHash { + if h := rawdb.FindCommonAncestor(c.chainDb.BlockStore(), prevHeader, header); h != nil { c.newHead(h.Number.Uint64(), true) } } diff --git a/core/genesis.go b/core/genesis.go index f1b4490f2e..262876b123 100644 --- a/core/genesis.go +++ b/core/genesis.go @@ -502,13 +502,13 @@ func (g *Genesis) Commit(db ethdb.Database, triedb *triedb.Database) (*types.Blo if err := flushAlloc(&g.Alloc, db, triedb, block.Hash()); err != nil { return nil, err } - rawdb.WriteTd(db, block.Hash(), block.NumberU64(), block.Difficulty()) - rawdb.WriteBlock(db, block) - rawdb.WriteReceipts(db, block.Hash(), block.NumberU64(), nil) - rawdb.WriteCanonicalHash(db, block.Hash(), block.NumberU64()) - rawdb.WriteHeadBlockHash(db, block.Hash()) + rawdb.WriteTd(db.BlockStore(), block.Hash(), block.NumberU64(), block.Difficulty()) + rawdb.WriteBlock(db.BlockStore(), block) + rawdb.WriteReceipts(db.BlockStore(), block.Hash(), block.NumberU64(), nil) + rawdb.WriteCanonicalHash(db.BlockStore(), block.Hash(), block.NumberU64()) + rawdb.WriteHeadBlockHash(db.BlockStore(), block.Hash()) rawdb.WriteHeadFastBlockHash(db, block.Hash()) - rawdb.WriteHeadHeaderHash(db, block.Hash()) + rawdb.WriteHeadHeaderHash(db.BlockStore(), block.Hash()) rawdb.WriteChainConfig(db, block.Hash(), config) return block, nil } diff --git a/core/headerchain.go b/core/headerchain.go index 9ce8d11c40..43e09b98a0 100644 --- a/core/headerchain.go +++ b/core/headerchain.go @@ -89,7 +89,7 @@ func NewHeaderChain(chainDb ethdb.Database, config *params.ChainConfig, engine c return nil, ErrNoGenesis } hc.currentHeader.Store(hc.genesisHeader) - if head := rawdb.ReadHeadBlockHash(chainDb); head != (common.Hash{}) { + if head := rawdb.ReadHeadBlockHash(chainDb.BlockStore()); head != (common.Hash{}) { if chead := hc.GetHeaderByHash(head); chead != nil { hc.currentHeader.Store(chead) } @@ -105,7 +105,7 @@ func (hc *HeaderChain) GetBlockNumber(hash common.Hash) *uint64 { if cached, ok := hc.numberCache.Get(hash); ok { return &cached } - number := rawdb.ReadHeaderNumber(hc.chainDb, hash) + number := rawdb.ReadHeaderNumber(hc.chainDb.BlockStore(), hash) if number != nil { hc.numberCache.Add(hash, *number) } @@ -133,9 +133,9 @@ func (hc *HeaderChain) Reorg(headers []*types.Header) error { // pile them onto the existing chain. Otherwise, do the necessary // reorgs. var ( - first = headers[0] - last = headers[len(headers)-1] - batch = hc.chainDb.NewBatch() + first = headers[0] + last = headers[len(headers)-1] + blockBatch = hc.chainDb.BlockStore().NewBatch() ) if first.ParentHash != hc.currentHeaderHash { // Delete any canonical number assignments above the new head @@ -144,7 +144,7 @@ func (hc *HeaderChain) Reorg(headers []*types.Header) error { if hash == (common.Hash{}) { break } - rawdb.DeleteCanonicalHash(batch, i) + rawdb.DeleteCanonicalHash(blockBatch, i) } // Overwrite any stale canonical number assignments, going // backwards from the first header in this import until the @@ -155,7 +155,7 @@ func (hc *HeaderChain) Reorg(headers []*types.Header) error { headHash = header.Hash() ) for rawdb.ReadCanonicalHash(hc.chainDb, headNumber) != headHash { - rawdb.WriteCanonicalHash(batch, headHash, headNumber) + rawdb.WriteCanonicalHash(blockBatch, headHash, headNumber) if headNumber == 0 { break // It shouldn't be reached } @@ -170,16 +170,16 @@ func (hc *HeaderChain) Reorg(headers []*types.Header) error { for i := 0; i < len(headers)-1; i++ { hash := headers[i+1].ParentHash // Save some extra hashing num := headers[i].Number.Uint64() - rawdb.WriteCanonicalHash(batch, hash, num) - rawdb.WriteHeadHeaderHash(batch, hash) + rawdb.WriteCanonicalHash(blockBatch, hash, num) + rawdb.WriteHeadHeaderHash(blockBatch, hash) } // Write the last header hash := headers[len(headers)-1].Hash() num := headers[len(headers)-1].Number.Uint64() - rawdb.WriteCanonicalHash(batch, hash, num) - rawdb.WriteHeadHeaderHash(batch, hash) + rawdb.WriteCanonicalHash(blockBatch, hash, num) + rawdb.WriteHeadHeaderHash(blockBatch, hash) - if err := batch.Write(); err != nil { + if err := blockBatch.Write(); err != nil { return err } // Last step update all in-memory head header markers @@ -205,7 +205,7 @@ func (hc *HeaderChain) WriteHeaders(headers []*types.Header) (int, error) { newTD = new(big.Int).Set(ptd) // Total difficulty of inserted chain inserted []rawdb.NumberHash // Ephemeral lookup of number/hash for the chain parentKnown = true // Set to true to force hc.HasHeader check the first iteration - batch = hc.chainDb.NewBatch() + blockBatch = hc.chainDb.BlockStore().NewBatch() ) for i, header := range headers { var hash common.Hash @@ -225,10 +225,10 @@ func (hc *HeaderChain) WriteHeaders(headers []*types.Header) (int, error) { alreadyKnown := parentKnown && hc.HasHeader(hash, number) if !alreadyKnown { // Irrelevant of the canonical status, write the TD and header to the database. - rawdb.WriteTd(batch, hash, number, newTD) + rawdb.WriteTd(blockBatch, hash, number, newTD) hc.tdCache.Add(hash, new(big.Int).Set(newTD)) - rawdb.WriteHeader(batch, header) + rawdb.WriteHeader(blockBatch, header) inserted = append(inserted, rawdb.NumberHash{Number: number, Hash: hash}) hc.headerCache.Add(hash, header) hc.numberCache.Add(hash, number) @@ -241,7 +241,7 @@ func (hc *HeaderChain) WriteHeaders(headers []*types.Header) (int, error) { return 0, errors.New("aborted") } // Commit to disk! - if err := batch.Write(); err != nil { + if err := blockBatch.Write(); err != nil { log.Crit("Failed to write headers", "error", err) } return len(inserted), nil @@ -562,7 +562,7 @@ func (hc *HeaderChain) setHead(headBlock uint64, headTime uint64, updateFn Updat } var ( parentHash common.Hash - batch = hc.chainDb.NewBatch() + blockBatch = hc.chainDb.BlockStore().NewBatch() origin = true ) done := func(header *types.Header) bool { @@ -597,7 +597,7 @@ func (hc *HeaderChain) setHead(headBlock uint64, headTime uint64, updateFn Updat } } // Update head header then. - rawdb.WriteHeadHeaderHash(markerBatch, parentHash) + rawdb.WriteHeadHeaderHash(hc.chainDb.BlockStore(), parentHash) if err := markerBatch.Write(); err != nil { log.Crit("Failed to update chain markers", "error", err) } @@ -626,16 +626,16 @@ func (hc *HeaderChain) setHead(headBlock uint64, headTime uint64, updateFn Updat } for _, hash := range hashes { if delFn != nil { - delFn(batch, hash, num) + delFn(blockBatch, hash, num) } - rawdb.DeleteHeader(batch, hash, num) - rawdb.DeleteTd(batch, hash, num) + rawdb.DeleteHeader(blockBatch, hash, num) + rawdb.DeleteTd(blockBatch, hash, num) } - rawdb.DeleteCanonicalHash(batch, num) + rawdb.DeleteCanonicalHash(blockBatch, num) } } // Flush all accumulated deletions. - if err := batch.Write(); err != nil { + if err := blockBatch.Write(); err != nil { log.Crit("Failed to rewind block", "error", err) } // Clear out any stale content from the caches diff --git a/core/rawdb/accessors_chain.go b/core/rawdb/accessors_chain.go index 025be7ade7..99091e420a 100644 --- a/core/rawdb/accessors_chain.go +++ b/core/rawdb/accessors_chain.go @@ -37,11 +37,11 @@ import ( // ReadCanonicalHash retrieves the hash assigned to a canonical block number. func ReadCanonicalHash(db ethdb.Reader, number uint64) common.Hash { var data []byte - db.ReadAncients(func(reader ethdb.AncientReaderOp) error { + db.BlockStoreReader().ReadAncients(func(reader ethdb.AncientReaderOp) error { data, _ = reader.Ancient(ChainFreezerHashTable, number) if len(data) == 0 { // Get it by hash from leveldb - data, _ = db.Get(headerHashKey(number)) + data, _ = db.BlockStoreReader().Get(headerHashKey(number)) } return nil }) @@ -303,7 +303,7 @@ func ReadHeaderRange(db ethdb.Reader, number uint64, count uint64) []rlp.RawValu // If we need to read live blocks, we need to figure out the hash first hash := ReadCanonicalHash(db, number) for ; i >= limit && count > 0; i-- { - if data, _ := db.Get(headerKey(i, hash)); len(data) > 0 { + if data, _ := db.BlockStoreReader().Get(headerKey(i, hash)); len(data) > 0 { rlpHeaders = append(rlpHeaders, data) // Get the parent hash for next query hash = types.HeaderParentHashFromRLP(data) @@ -336,7 +336,7 @@ func ReadHeaderRange(db ethdb.Reader, number uint64, count uint64) []rlp.RawValu // ReadHeaderRLP retrieves a block header in its raw RLP database encoding. func ReadHeaderRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue { var data []byte - db.ReadAncients(func(reader ethdb.AncientReaderOp) error { + db.BlockStoreReader().ReadAncients(func(reader ethdb.AncientReaderOp) error { // First try to look up the data in ancient database. Extra hash // comparison is necessary since ancient database only maintains // the canonical data. @@ -345,7 +345,7 @@ func ReadHeaderRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValu return nil } // If not, try reading from leveldb - data, _ = db.Get(headerKey(number, hash)) + data, _ = db.BlockStoreReader().Get(headerKey(number, hash)) return nil }) return data @@ -353,10 +353,10 @@ func ReadHeaderRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValu // HasHeader verifies the existence of a block header corresponding to the hash. func HasHeader(db ethdb.Reader, hash common.Hash, number uint64) bool { - if isCanon(db, number, hash) { + if isCanon(db.BlockStoreReader(), number, hash) { return true } - if has, err := db.Has(headerKey(number, hash)); !has || err != nil { + if has, err := db.BlockStoreReader().Has(headerKey(number, hash)); !has || err != nil { return false } return true @@ -429,14 +429,14 @@ func ReadBodyRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue // comparison is necessary since ancient database only maintains // the canonical data. var data []byte - db.ReadAncients(func(reader ethdb.AncientReaderOp) error { + db.BlockStoreReader().ReadAncients(func(reader ethdb.AncientReaderOp) error { // Check if the data is in ancients if isCanon(reader, number, hash) { data, _ = reader.Ancient(ChainFreezerBodiesTable, number) return nil } // If not, try reading from leveldb - data, _ = db.Get(blockBodyKey(number, hash)) + data, _ = db.BlockStoreReader().Get(blockBodyKey(number, hash)) return nil }) return data @@ -446,7 +446,7 @@ func ReadBodyRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue // block at number, in RLP encoding. func ReadCanonicalBodyRLP(db ethdb.Reader, number uint64) rlp.RawValue { var data []byte - db.ReadAncients(func(reader ethdb.AncientReaderOp) error { + db.BlockStoreReader().ReadAncients(func(reader ethdb.AncientReaderOp) error { data, _ = reader.Ancient(ChainFreezerBodiesTable, number) if len(data) > 0 { return nil @@ -455,7 +455,7 @@ func ReadCanonicalBodyRLP(db ethdb.Reader, number uint64) rlp.RawValue { // Note: ReadCanonicalHash cannot be used here because it also // calls ReadAncients internally. hash, _ := db.Get(headerHashKey(number)) - data, _ = db.Get(blockBodyKey(number, common.BytesToHash(hash))) + data, _ = db.BlockStoreReader().Get(blockBodyKey(number, common.BytesToHash(hash))) return nil }) return data @@ -470,10 +470,10 @@ func WriteBodyRLP(db ethdb.KeyValueWriter, hash common.Hash, number uint64, rlp // HasBody verifies the existence of a block body corresponding to the hash. func HasBody(db ethdb.Reader, hash common.Hash, number uint64) bool { - if isCanon(db, number, hash) { + if isCanon(db.BlockStoreReader(), number, hash) { return true } - if has, err := db.Has(blockBodyKey(number, hash)); !has || err != nil { + if has, err := db.BlockStoreReader().Has(blockBodyKey(number, hash)); !has || err != nil { return false } return true @@ -512,14 +512,14 @@ func DeleteBody(db ethdb.KeyValueWriter, hash common.Hash, number uint64) { // ReadTdRLP retrieves a block's total difficulty corresponding to the hash in RLP encoding. func ReadTdRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue { var data []byte - db.ReadAncients(func(reader ethdb.AncientReaderOp) error { + db.BlockStoreReader().ReadAncients(func(reader ethdb.AncientReaderOp) error { // Check if the data is in ancients if isCanon(reader, number, hash) { data, _ = reader.Ancient(ChainFreezerDifficultyTable, number) return nil } // If not, try reading from leveldb - data, _ = db.Get(headerTDKey(number, hash)) + data, _ = db.BlockStoreReader().Get(headerTDKey(number, hash)) return nil }) return data @@ -560,10 +560,10 @@ func DeleteTd(db ethdb.KeyValueWriter, hash common.Hash, number uint64) { // HasReceipts verifies the existence of all the transaction receipts belonging // to a block. func HasReceipts(db ethdb.Reader, hash common.Hash, number uint64) bool { - if isCanon(db, number, hash) { + if isCanon(db.BlockStoreReader(), number, hash) { return true } - if has, err := db.Has(blockReceiptsKey(number, hash)); !has || err != nil { + if has, err := db.BlockStoreReader().Has(blockReceiptsKey(number, hash)); !has || err != nil { return false } return true @@ -572,14 +572,14 @@ func HasReceipts(db ethdb.Reader, hash common.Hash, number uint64) bool { // ReadReceiptsRLP retrieves all the transaction receipts belonging to a block in RLP encoding. func ReadReceiptsRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue { var data []byte - db.ReadAncients(func(reader ethdb.AncientReaderOp) error { + db.BlockStoreReader().ReadAncients(func(reader ethdb.AncientReaderOp) error { // Check if the data is in ancients if isCanon(reader, number, hash) { data, _ = reader.Ancient(ChainFreezerReceiptTable, number) return nil } // If not, try reading from leveldb - data, _ = db.Get(blockReceiptsKey(number, hash)) + data, _ = db.BlockStoreReader().Get(blockReceiptsKey(number, hash)) return nil }) return data @@ -950,24 +950,24 @@ func FindCommonAncestor(db ethdb.Reader, a, b *types.Header) *types.Header { // ReadHeadHeader returns the current canonical head header. func ReadHeadHeader(db ethdb.Reader) *types.Header { - headHeaderHash := ReadHeadHeaderHash(db) + headHeaderHash := ReadHeadHeaderHash(db.BlockStoreReader()) if headHeaderHash == (common.Hash{}) { return nil } - headHeaderNumber := ReadHeaderNumber(db, headHeaderHash) + headHeaderNumber := ReadHeaderNumber(db.BlockStoreReader(), headHeaderHash) if headHeaderNumber == nil { return nil } - return ReadHeader(db, headHeaderHash, *headHeaderNumber) + return ReadHeader(db.BlockStoreReader(), headHeaderHash, *headHeaderNumber) } // ReadHeadBlock returns the current canonical head block. func ReadHeadBlock(db ethdb.Reader) *types.Block { - headBlockHash := ReadHeadBlockHash(db) + headBlockHash := ReadHeadBlockHash(db.BlockStoreReader()) if headBlockHash == (common.Hash{}) { return nil } - headBlockNumber := ReadHeaderNumber(db, headBlockHash) + headBlockNumber := ReadHeaderNumber(db.BlockStoreReader(), headBlockHash) if headBlockNumber == nil { return nil } diff --git a/core/rawdb/accessors_indexes.go b/core/rawdb/accessors_indexes.go index 4f2ef0a880..95fc1d7806 100644 --- a/core/rawdb/accessors_indexes.go +++ b/core/rawdb/accessors_indexes.go @@ -42,7 +42,7 @@ func ReadTxLookupEntry(db ethdb.Reader, hash common.Hash) *uint64 { } // Database v4-v5 tx lookup format just stores the hash if len(data) == common.HashLength { - return ReadHeaderNumber(db, common.BytesToHash(data)) + return ReadHeaderNumber(db.BlockStoreReader(), common.BytesToHash(data)) } // Finally try database v3 tx lookup format var entry LegacyTxLookupEntry diff --git a/core/rawdb/ancient_utils.go b/core/rawdb/ancient_utils.go index 6ee8909987..170be10bc2 100644 --- a/core/rawdb/ancient_utils.go +++ b/core/rawdb/ancient_utils.go @@ -119,16 +119,25 @@ func inspectFreezers(db ethdb.Database) ([]freezerInfo, error) { // ancient indicates the path of root ancient directory where the chain freezer can // be opened. Start and end specify the range for dumping out indexes. // Note this function can only be used for debugging purposes. -func InspectFreezerTable(ancient string, freezerName string, tableName string, start, end int64) error { +func InspectFreezerTable(ancient string, freezerName string, tableName string, start, end int64, multiDatabase bool) error { var ( path string tables map[string]bool ) switch freezerName { case ChainFreezerName: - path, tables = resolveChainFreezerDir(ancient), chainFreezerNoSnappy + if multiDatabase { + path, tables = resolveChainFreezerDir(filepath.Dir(ancient)+"/block/ancient"), chainFreezerNoSnappy + } else { + path, tables = resolveChainFreezerDir(ancient), chainFreezerNoSnappy + } + case StateFreezerName: - path, tables = filepath.Join(ancient, freezerName), stateFreezerNoSnappy + if multiDatabase { + path, tables = filepath.Join(filepath.Dir(ancient)+"/state/ancient", freezerName), stateFreezerNoSnappy + } else { + path, tables = filepath.Join(ancient, freezerName), stateFreezerNoSnappy + } default: return fmt.Errorf("unknown freezer, supported ones: %v", freezers) } diff --git a/core/rawdb/chain_iterator.go b/core/rawdb/chain_iterator.go index 759e5913d1..930b479623 100644 --- a/core/rawdb/chain_iterator.go +++ b/core/rawdb/chain_iterator.go @@ -78,7 +78,7 @@ func InitDatabaseFromFreezer(db ethdb.Database) { } batch.Reset() - WriteHeadHeaderHash(db, hash) + WriteHeadHeaderHash(db.BlockStore(), hash) WriteHeadFastBlockHash(db, hash) log.Info("Initialized database from freezer", "blocks", frozen, "elapsed", common.PrettyDuration(time.Since(start))) } @@ -117,7 +117,7 @@ func iterateTransactions(db ethdb.Database, from uint64, to uint64, reverse bool } defer close(rlpCh) for n != end { - data := ReadCanonicalBodyRLP(db, n) + data := ReadCanonicalBodyRLP(db.BlockStore(), n) // Feed the block to the aggregator, or abort on interrupt select { case rlpCh <- &numberRlp{n, data}: diff --git a/core/rawdb/database.go b/core/rawdb/database.go index 3a50a0d33d..cfd9345c03 100644 --- a/core/rawdb/database.go +++ b/core/rawdb/database.go @@ -42,6 +42,7 @@ type freezerdb struct { readOnly bool ancientRoot string stateStore ethdb.Database + blockStore ethdb.Database } func (frdb *freezerdb) StateStoreReader() ethdb.Reader { @@ -51,6 +52,18 @@ func (frdb *freezerdb) StateStoreReader() ethdb.Reader { return frdb.stateStore } +func (frdb *freezerdb) BlockStoreReader() ethdb.Reader { + if frdb.blockStore == nil { + return frdb + } + return frdb.blockStore +} + +func (frdb *freezerdb) BlockStoreWriter() ethdb.Writer { + //TODO implement me + panic("implement me") +} + // AncientDatadir returns the path of root ancient directory. func (frdb *freezerdb) AncientDatadir() (string, error) { return frdb.ancientRoot, nil @@ -72,6 +85,11 @@ func (frdb *freezerdb) Close() error { errs = append(errs, err) } } + if frdb.blockStore != nil { + if err := frdb.blockStore.Close(); err != nil { + errs = append(errs, err) + } + } if len(errs) != 0 { return fmt.Errorf("%v", errs) } @@ -89,6 +107,21 @@ func (frdb *freezerdb) SetStateStore(state ethdb.Database) { frdb.stateStore = state } +func (frdb *freezerdb) BlockStore() ethdb.Database { + if frdb.blockStore != nil { + return frdb.blockStore + } else { + return frdb + } +} + +func (frdb *freezerdb) SetBlockStore(block ethdb.Database) { + if frdb.blockStore != nil { + frdb.blockStore.Close() + } + frdb.blockStore = block +} + // Freeze is a helper method used for external testing to trigger and block until // a freeze cycle completes, without having to sleep for a minute to trigger the // automatic background run. @@ -107,6 +140,7 @@ func (frdb *freezerdb) Freeze() error { type nofreezedb struct { ethdb.KeyValueStore stateStore ethdb.Database + blockStore ethdb.Database } // HasAncient returns an error as we don't have a backing chain freezer. @@ -174,6 +208,31 @@ func (db *nofreezedb) StateStoreReader() ethdb.Reader { return db } +func (db *nofreezedb) BlockStore() ethdb.Database { + if db.blockStore != nil { + return db.blockStore + } + return db +} + +func (db *nofreezedb) SetBlockStore(block ethdb.Database) { + db.blockStore = block +} + +func (db *nofreezedb) BlockStoreReader() ethdb.Reader { + if db.blockStore != nil { + return db.blockStore + } + return db +} + +func (db *nofreezedb) BlockStoreWriter() ethdb.Writer { + if db.blockStore != nil { + return db.blockStore + } + return db +} + func (db *nofreezedb) ReadAncients(fn func(reader ethdb.AncientReaderOp) error) (err error) { // Unlike other ancient-related methods, this method does not return // errNotSupported when invoked. @@ -491,6 +550,48 @@ func (s *stat) Count() string { return s.count.String() } +type DataType int + +const ( + StateDataType DataType = iota + BlockDataType + ChainDataType + Unknown +) + +func DataTypeByKey(key []byte) DataType { + switch { + // state + case IsLegacyTrieNode(key, key), + bytes.HasPrefix(key, stateIDPrefix) && len(key) == len(stateIDPrefix)+common.HashLength, + IsAccountTrieNode(key), + IsStorageTrieNode(key): + return StateDataType + + // block + case bytes.HasPrefix(key, headerPrefix) && len(key) == (len(headerPrefix)+8+common.HashLength), + bytes.HasPrefix(key, blockBodyPrefix) && len(key) == (len(blockBodyPrefix)+8+common.HashLength), + bytes.HasPrefix(key, blockReceiptsPrefix) && len(key) == (len(blockReceiptsPrefix)+8+common.HashLength), + bytes.HasPrefix(key, headerPrefix) && bytes.HasSuffix(key, headerTDSuffix), + bytes.HasPrefix(key, headerPrefix) && bytes.HasSuffix(key, headerHashSuffix), + bytes.HasPrefix(key, headerNumberPrefix) && len(key) == (len(headerNumberPrefix)+common.HashLength): + return BlockDataType + default: + for _, meta := range [][]byte{ + fastTrieProgressKey, persistentStateIDKey, trieJournalKey, snapSyncStatusFlagKey} { + if bytes.Equal(key, meta) { + return StateDataType + } + } + for _, meta := range [][]byte{headHeaderKey, headFinalizedBlockKey} { + if bytes.Equal(key, meta) { + return BlockDataType + } + } + return ChainDataType + } +} + // InspectDatabase traverses the entire database and checks the size // of all different categories of data. func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error { @@ -498,10 +599,15 @@ func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error { defer it.Release() var trieIter ethdb.Iterator + var blockIter ethdb.Iterator if db.StateStore() != nil { trieIter = db.StateStore().NewIterator(keyPrefix, nil) defer trieIter.Release() } + if db.BlockStore() != db { + blockIter = db.BlockStore().NewIterator(keyPrefix, nil) + defer blockIter.Release() + } var ( count int64 start = time.Now() @@ -631,6 +737,7 @@ func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error { value = trieIter.Value() size = common.StorageSize(len(key) + len(value)) ) + total += size switch { case IsLegacyTrieNode(key, value): @@ -644,9 +751,10 @@ func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error { default: var accounted bool for _, meta := range [][]byte{ - fastTrieProgressKey, persistentStateIDKey, trieJournalKey} { + fastTrieProgressKey, persistentStateIDKey, trieJournalKey, snapSyncStatusFlagKey} { if bytes.Equal(key, meta) { metadata.Add(size) + accounted = true break } } @@ -660,6 +768,54 @@ func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error { logged = time.Now() } } + log.Info("Inspecting separate state database", "count", count, "elapsed", common.PrettyDuration(time.Since(start))) + } + // inspect separate block db + if blockIter != nil { + count = 0 + logged = time.Now() + + for blockIter.Next() { + var ( + key = blockIter.Key() + value = blockIter.Value() + size = common.StorageSize(len(key) + len(value)) + ) + total += size + + switch { + case bytes.HasPrefix(key, headerPrefix) && len(key) == (len(headerPrefix)+8+common.HashLength): + headers.Add(size) + case bytes.HasPrefix(key, blockBodyPrefix) && len(key) == (len(blockBodyPrefix)+8+common.HashLength): + bodies.Add(size) + case bytes.HasPrefix(key, blockReceiptsPrefix) && len(key) == (len(blockReceiptsPrefix)+8+common.HashLength): + receipts.Add(size) + case bytes.HasPrefix(key, headerPrefix) && bytes.HasSuffix(key, headerTDSuffix): + tds.Add(size) + case bytes.HasPrefix(key, headerPrefix) && bytes.HasSuffix(key, headerHashSuffix): + numHashPairings.Add(size) + case bytes.HasPrefix(key, headerNumberPrefix) && len(key) == (len(headerNumberPrefix)+common.HashLength): + hashNumPairings.Add(size) + default: + var accounted bool + for _, meta := range [][]byte{headHeaderKey, headFinalizedBlockKey} { + if bytes.Equal(key, meta) { + metadata.Add(size) + accounted = true + break + } + } + if !accounted { + unaccounted.Add(size) + } + } + count++ + if count%1000 == 0 && time.Since(logged) > 8*time.Second { + log.Info("Inspecting separate block database", "count", count, "elapsed", common.PrettyDuration(time.Since(start))) + logged = time.Now() + } + } + log.Info("Inspecting separate block database", "count", count, "elapsed", common.PrettyDuration(time.Since(start))) } // Display the database statistic of key-value store. stats := [][]string{ @@ -686,7 +842,7 @@ func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error { {"Light client", "Bloom trie nodes", bloomTrieNodes.Size(), bloomTrieNodes.Count()}, } // Inspect all registered append-only file store then. - ancients, err := inspectFreezers(db) + ancients, err := inspectFreezers(db.BlockStore()) if err != nil { return err } @@ -772,3 +928,26 @@ func ReadChainMetadata(db ethdb.KeyValueStore) [][]string { } return data } + +func ReadChainMetadataFromMultiDatabase(db ethdb.Database) [][]string { + pp := func(val *uint64) string { + if val == nil { + return "" + } + return fmt.Sprintf("%d (%#x)", *val, *val) + } + data := [][]string{ + {"databaseVersion", pp(ReadDatabaseVersion(db))}, + {"headBlockHash", fmt.Sprintf("%v", ReadHeadBlockHash(db.BlockStore()))}, + {"headFastBlockHash", fmt.Sprintf("%v", ReadHeadFastBlockHash(db))}, + {"headHeaderHash", fmt.Sprintf("%v", ReadHeadHeaderHash(db.BlockStore()))}, + {"lastPivotNumber", pp(ReadLastPivotNumber(db))}, + {"len(snapshotSyncStatus)", fmt.Sprintf("%d bytes", len(ReadSnapshotSyncStatus(db)))}, + {"snapshotDisabled", fmt.Sprintf("%v", ReadSnapshotDisabled(db))}, + {"snapshotJournal", fmt.Sprintf("%d bytes", len(ReadSnapshotJournal(db)))}, + {"snapshotRecoveryNumber", pp(ReadSnapshotRecoveryNumber(db))}, + {"snapshotRoot", fmt.Sprintf("%v", ReadSnapshotRoot(db))}, + {"txIndexTail", pp(ReadTxIndexTail(db))}, + } + return data +} diff --git a/core/rawdb/table.go b/core/rawdb/table.go index a945c3e905..a6233406ef 100644 --- a/core/rawdb/table.go +++ b/core/rawdb/table.go @@ -27,6 +27,22 @@ type table struct { prefix string } +func (t *table) BlockStoreReader() ethdb.Reader { + return t +} + +func (t *table) BlockStoreWriter() ethdb.Writer { + return t +} + +func (t *table) BlockStore() ethdb.Database { + return t +} + +func (t *table) SetBlockStore(block ethdb.Database) { + panic("not implement") +} + // NewTable returns a database object that prefixes all keys with a given string. func NewTable(db ethdb.Database, prefix string) ethdb.Database { return &table{ diff --git a/ethdb/database.go b/ethdb/database.go index 7fea031bf3..e1249d0a18 100644 --- a/ethdb/database.go +++ b/ethdb/database.go @@ -157,12 +157,26 @@ type StateStoreReader interface { StateStoreReader() Reader } +type BlockStore interface { + BlockStore() Database + SetBlockStore(block Database) +} + +type BlockStoreReader interface { + BlockStoreReader() Reader +} + +type BlockStoreWriter interface { + BlockStoreWriter() Writer +} + // Reader contains the methods required to read data from both key-value as well as // immutable ancient data. type Reader interface { KeyValueReader AncientReader StateStoreReader + BlockStoreReader } // Writer contains the methods required to write data to both key-value as well as @@ -170,6 +184,7 @@ type Reader interface { type Writer interface { KeyValueWriter AncientWriter + BlockStoreWriter } // Stater contains the methods required to retrieve states from both key-value as well as @@ -206,6 +221,7 @@ type Database interface { Reader Writer StateStore + BlockStore Batcher Iteratee Stater diff --git a/ethdb/remotedb/remotedb.go b/ethdb/remotedb/remotedb.go index e4df060108..705816ad93 100644 --- a/ethdb/remotedb/remotedb.go +++ b/ethdb/remotedb/remotedb.go @@ -32,6 +32,22 @@ type Database struct { remote *rpc.Client } +func (db *Database) BlockStoreReader() ethdb.Reader { + return db +} + +func (db *Database) BlockStoreWriter() ethdb.Writer { + return db +} + +func (db *Database) BlockStore() ethdb.Database { + return db +} + +func (db *Database) SetBlockStore(block ethdb.Database) { + panic("not supported") +} + func (db *Database) Has(key []byte) (bool, error) { if _, err := db.Get(key); err != nil { return false, nil diff --git a/node/errors.go b/node/errors.go index 67547bf691..e7f60ce9bc 100644 --- a/node/errors.go +++ b/node/errors.go @@ -24,10 +24,11 @@ import ( ) var ( - ErrDatadirUsed = errors.New("datadir already used by another process") - ErrNodeStopped = errors.New("node not started") - ErrNodeRunning = errors.New("node already running") - ErrServiceUnknown = errors.New("unknown service") + ErrDatadirUsed = errors.New("datadir already used by another process") + ErrNodeStopped = errors.New("node not started") + ErrNodeRunning = errors.New("node already running") + ErrServiceUnknown = errors.New("unknown service") + ErrSeprateDBDatadir = errors.New("datadir is not configured when using separate trie") datadirInUseErrnos = map[uint]bool{11: true, 32: true, 35: true} ) diff --git a/node/node.go b/node/node.go index 7b0e945fae..35d2705cf6 100644 --- a/node/node.go +++ b/node/node.go @@ -73,6 +73,12 @@ const ( initializingState = iota runningState closedState + blockDbCacheSize = 256 + blockDbHandlesMinSize = 1000 + blockDbHandlesMaxSize = 2000 + chainDbMemoryPercentage = 50 + chainDbHandlesPercentage + diffStoreHandlesPercentage = 20 ) // New creates a new P2P node, ready for protocol registration. @@ -740,21 +746,43 @@ func (n *Node) OpenDatabase(name string, cache, handles int, namespace string, r } func (n *Node) OpenAndMergeDatabase(name string, cache, handles int, ancient, namespace string, readonly bool) (ethdb.Database, error) { - chainDataHandles := handles + var ( + err error + stateDiskDb ethdb.Database + blockDb ethdb.Database + blockDbHandlesSize int + chainDataHandles = handles + chainDbCache = cache + ) - var statediskdb ethdb.Database - var err error + isMultiDatabase := n.CheckIfMultiDataBase() // Open the separated state database if the state directory exists - if n.IsSeparatedDB() { - // Allocate half of the handles and cache to this separate state data database - statediskdb, err = n.OpenDatabaseWithFreezer(name+"/state", cache/2, chainDataHandles/2, "", "eth/db/statedata/", readonly) + if isMultiDatabase { + // Resource allocation rules: + // 1) Allocate a fixed percentage of memory for chainDb based on chainDbMemoryPercentage & chainDbHandlesPercentage. + // 2) Allocate a fixed size for blockDb based on blockDbCacheSize & blockDbHandlesSize. + // 3) Allocate the remaining resources to stateDb. + chainDbCache = int(float64(cache) * chainDbMemoryPercentage / 100) + chainDataHandles = int(float64(handles) * chainDbHandlesPercentage / 100) + if handles/10 > blockDbHandlesMaxSize { + blockDbHandlesSize = blockDbHandlesMaxSize + } else { + blockDbHandlesSize = blockDbHandlesMinSize + } + stateDbCache := cache - chainDbCache - blockDbCacheSize + stateDbHandles := handles - chainDataHandles - blockDbHandlesSize + + // Allocate half of the handles and chainDbCache to this separate state data database + stateDiskDb, err = n.OpenDatabaseWithFreezer(name+"/state", stateDbCache, stateDbHandles, "", "eth/db/statedata/", readonly) if err != nil { return nil, err } - // Reduce the handles and cache to this separate database because it is not a complete database with no trie data storing in it. - cache = int(float64(cache) * 0.6) - chainDataHandles = int(float64(chainDataHandles) * 0.6) + blockDb, err = n.OpenDatabaseWithFreezer(name+"/block", blockDbCacheSize, blockDbHandlesSize, "", "eth/db/blockdata/", readonly) + if err != nil { + return nil, err + } + log.Warn("Multi-database is an experimental feature") } chainDB, err := n.OpenDatabaseWithFreezer(name, cache, chainDataHandles, ancient, namespace, readonly) @@ -762,8 +790,9 @@ func (n *Node) OpenAndMergeDatabase(name string, cache, handles int, ancient, na return nil, err } - if statediskdb != nil { - chainDB.SetStateStore(statediskdb) + if isMultiDatabase { + chainDB.SetStateStore(stateDiskDb) + chainDB.SetBlockStore(blockDb) } return chainDB, nil @@ -802,14 +831,31 @@ func (n *Node) OpenDatabaseWithFreezer(name string, cache, handles int, ancient return db, err } -// IsSeparatedDB check the state subdirectory of db, if subdirectory exists, return true -func (n *Node) IsSeparatedDB() bool { - separateDir := filepath.Join(n.ResolvePath("chaindata"), "state") - fileInfo, err := os.Stat(separateDir) - if os.IsNotExist(err) { - return false +// CheckIfMultiDataBase check the state and block subdirectory of db, if subdirectory exists, return true +func (n *Node) CheckIfMultiDataBase() bool { + var ( + stateExist = true + blockExist = true + ) + + separateStateDir := filepath.Join(n.ResolvePath("chaindata"), "state") + fileInfo, stateErr := os.Stat(separateStateDir) + if os.IsNotExist(stateErr) || !fileInfo.IsDir() { + stateExist = false + } + separateBlockDir := filepath.Join(n.ResolvePath("chaindata"), "block") + blockFileInfo, blockErr := os.Stat(separateBlockDir) + if os.IsNotExist(blockErr) || !blockFileInfo.IsDir() { + blockExist = false + } + + if stateExist && blockExist { + return true + } else if !stateExist && !blockExist { + return false + } else { + panic("data corruption! missing block or state dir.") } - return fileInfo.IsDir() } // ResolvePath returns the absolute path of a resource in the instance directory.