From 1d713ac1afc273d6ab9f2dcd34eef20a59691b5c Mon Sep 17 00:00:00 2001 From: Fynn Date: Tue, 21 Nov 2023 11:22:15 +0800 Subject: [PATCH] cmd/geth: add hbss2pbss tool --- cmd/geth/dbcmd.go | 112 +++++++++++++++++ cmd/utils/flags.go | 6 + core/rawdb/ancient_utils.go | 23 ++++ core/rawdb/database.go | 23 ++++ core/rawdb/freezer_table.go | 30 +++++ trie/hbss2pbss.go | 243 ++++++++++++++++++++++++++++++++++++ 6 files changed, 437 insertions(+) create mode 100644 trie/hbss2pbss.go diff --git a/cmd/geth/dbcmd.go b/cmd/geth/dbcmd.go index 6f802716c5..3ffdb2eddb 100644 --- a/cmd/geth/dbcmd.go +++ b/cmd/geth/dbcmd.go @@ -19,6 +19,7 @@ package main import ( "bytes" "fmt" + "math" "os" "os/signal" "path/filepath" @@ -133,6 +134,19 @@ corruption if it is aborted during execution'!`, Description: `This command deletes the specified database key from the database. WARNING: This is a low-level operation which may cause database corruption!`, } + dbHbss2PbssCmd = &cli.Command{ + Action: hbss2pbss, + Name: "hbss-to-pbss", + ArgsUsage: "", + Flags: []cli.Flag{ + utils.DataDirFlag, + utils.SyncModeFlag, + utils.ForceFlag, + utils.AncientFlag, + }, + Usage: "Convert Hash-Base to Path-Base trie node.", + Description: `This command iterates the entire trie node database and convert the hash-base node to path-base node.`, + } dbPutCmd = &cli.Command{ Action: dbPut, Name: "put", @@ -724,3 +738,101 @@ func showMetaData(ctx *cli.Context) error { table.Render() return nil } + +func hbss2pbss(ctx *cli.Context) error { + if ctx.NArg() > 1 { + return fmt.Errorf("required arguments: %v", ctx.Command.ArgsUsage) + } + + var jobnum uint64 + var err error + if ctx.NArg() == 1 { + jobnum, err = strconv.ParseUint(ctx.Args().Get(0), 10, 64) + if err != nil { + return fmt.Errorf("failed to Parse jobnum, Args[1]: %v, err: %v", ctx.Args().Get(1), err) + } + } else { + // by default + jobnum = 1000 + } + + force := ctx.Bool(utils.ForceFlag.Name) + + stack, _ := makeConfigNode(ctx) + defer stack.Close() + + db := utils.MakeChainDatabase(ctx, stack, false) + db.Sync() + defer db.Close() + + // convert hbss trie node to pbss trie node + lastStateID := rawdb.ReadPersistentStateID(db) + if lastStateID == 0 || force { + config := trie.HashDefaults + triedb := trie.NewDatabase(db, config) + triedb.Cap(0) + log.Info("hbss2pbss triedb", "scheme", triedb.Scheme()) + defer triedb.Close() + + headerHash := rawdb.ReadHeadHeaderHash(db) + blockNumber := rawdb.ReadHeaderNumber(db, headerHash) + if blockNumber == nil { + log.Error("read header number failed.") + return fmt.Errorf("read header number failed") + } + + log.Info("hbss2pbss converting", "HeaderHash: ", headerHash.String(), ", blockNumber: ", *blockNumber) + + var headerBlockHash common.Hash + var trieRootHash common.Hash + + if *blockNumber != math.MaxUint64 { + headerBlockHash = rawdb.ReadCanonicalHash(db, *blockNumber) + if headerBlockHash == (common.Hash{}) { + return fmt.Errorf("ReadHeadBlockHash empty hash") + } + blockHeader := rawdb.ReadHeader(db, headerBlockHash, *blockNumber) + trieRootHash = blockHeader.Root + fmt.Println("Canonical Hash: ", headerBlockHash.String(), ", TrieRootHash: ", trieRootHash.String()) + } + if (trieRootHash == common.Hash{}) { + log.Error("Empty root hash") + return fmt.Errorf("Empty root hash.") + } + + id := trie.StateTrieID(trieRootHash) + theTrie, err := trie.New(id, triedb) + if err != nil { + log.Error("fail to new trie tree", "err", err, "rootHash", err, trieRootHash.String()) + return err + } + + h2p, err := trie.NewHbss2Pbss(theTrie, triedb, trieRootHash, *blockNumber, jobnum) + if err != nil { + log.Error("fail to new hash2pbss", "err", err, "rootHash", err, trieRootHash.String()) + return err + } + h2p.Run() + } else { + log.Info("Convert hbss to pbss success. Nothing to do.") + } + + // repair state ancient offset + lastStateID = rawdb.ReadPersistentStateID(db) + if lastStateID == 0 { + log.Error("Convert hbss to pbss trie node error. The last state id is still 0") + } + ancient := stack.ResolveAncient("chaindata", ctx.String(utils.AncientFlag.Name)) + err = rawdb.ResetStateFreezerTableOffset(ancient, lastStateID) + if err != nil { + log.Error("Reset state freezer table offset failed", "error", err) + return err + } + // prune hbss trie node + err = rawdb.PruneHashTrieNodeInDataBase(db) + if err != nil { + log.Error("Prune Hash trie node in database failed", "error", err) + return err + } + return nil +} diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index db226c73d8..225836e609 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -259,6 +259,12 @@ var ( Value: &defaultSyncMode, Category: flags.StateCategory, } + // hbss2pbss command options + ForceFlag = &cli.BoolFlag{ + Name: "force", + Usage: "Force convert hbss trie node to pbss trie node. Ingore any metadata", + Value: false, + } GCModeFlag = &cli.StringFlag{ Name: "gcmode", Usage: `Blockchain garbage collection mode, only relevant in state.scheme=hash ("full", "archive")`, diff --git a/core/rawdb/ancient_utils.go b/core/rawdb/ancient_utils.go index dfb2fdfb67..7c8eebe85e 100644 --- a/core/rawdb/ancient_utils.go +++ b/core/rawdb/ancient_utils.go @@ -18,9 +18,12 @@ package rawdb import ( "fmt" + "path/filepath" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/metrics" ) type tableSize struct { @@ -144,3 +147,23 @@ func InspectFreezerTable(ancient string, freezerName string, tableName string, s table.dumpIndexStdout(start, end) return nil } + +func ResetStateFreezerTableOffset(ancient string, virtualTail uint64) error { + path, tables := filepath.Join(ancient, stateFreezerName), stateFreezerNoSnappy + + for name, disableSnappy := range tables { + log.Info("Handle table", "name", name, "disableSnappy", disableSnappy) + table, err := newTable(path, name, metrics.NilMeter{}, metrics.NilMeter{}, metrics.NilGauge{}, freezerTableSize, disableSnappy, false) + if err != nil { + log.Error("New table failed", "error", err) + return err + } + // Reset the metadata of the freezer table + err = table.ResetItemsOffset(virtualTail) + if err != nil { + log.Error("Reset items offset of the table", "name", name, "error", err) + return err + } + } + return nil +} diff --git a/core/rawdb/database.go b/core/rawdb/database.go index 1d7b7d1ca8..b0b3d8b554 100644 --- a/core/rawdb/database.go +++ b/core/rawdb/database.go @@ -664,3 +664,26 @@ func ReadChainMetadata(db ethdb.KeyValueStore) [][]string { } return data } + +// PruneHashTrieNodeInDataBase prune all hash trie node +func PruneHashTrieNodeInDataBase(db ethdb.Database) error { + it := db.NewIterator([]byte{}, []byte{}) + defer it.Release() + + total_num := 0 + for it.Next() { + var key = it.Key() + switch { + case IsLegacyTrieNode(key, it.Value()): + db.Delete(key) + total_num++ + if total_num%100000 == 0 { + log.Info("Pruning hash-base state trie nodes", "Complete progress: ", total_num) + } + default: + continue + } + } + log.Info("Pruning hash-base state trie nodes", "Complete progress", total_num) + return nil +} diff --git a/core/rawdb/freezer_table.go b/core/rawdb/freezer_table.go index e3353cc7d5..4eba556fb2 100644 --- a/core/rawdb/freezer_table.go +++ b/core/rawdb/freezer_table.go @@ -958,3 +958,33 @@ func (t *freezerTable) dumpIndex(w io.Writer, start, stop int64) { } fmt.Fprintf(w, "|--------------------------|\n") } + +func (t *freezerTable) ResetItemsOffset(virtualTail uint64) error { + stat, err := t.index.Stat() + if err != nil { + return err + } + + if stat.Size() == 0 { + return fmt.Errorf("Stat size is zero when ResetVirtualTail.") + } + + var firstIndex indexEntry + + buffer := make([]byte, indexEntrySize) + + t.index.ReadAt(buffer, 0) + firstIndex.unmarshalBinary(buffer) + + firstIndex.offset = uint32(virtualTail) + t.index.WriteAt(firstIndex.append(nil), 0) + + var firstIndex2 indexEntry + buffer2 := make([]byte, indexEntrySize) + t.index.ReadAt(buffer2, 0) + firstIndex2.unmarshalBinary(buffer2) + + log.Info("Reset Index", "filenum", t.index.Name(), "offset", firstIndex2.offset) + + return nil +} diff --git a/trie/hbss2pbss.go b/trie/hbss2pbss.go new file mode 100644 index 0000000000..1faaf34783 --- /dev/null +++ b/trie/hbss2pbss.go @@ -0,0 +1,243 @@ +package trie + +import ( + "bytes" + "errors" + "fmt" + "runtime" + "sync" + "sync/atomic" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/trie/trienode" +) + +type Hbss2Pbss struct { + trie *Trie // traverse trie + db *Database + blocknum uint64 + root node // root of triedb + stateRootHash common.Hash + concurrentQueue chan struct{} + totalNum uint64 + wg sync.WaitGroup +} + +const ( + DEFAULT_TRIEDBCACHE_SIZE = 1024 * 1024 * 1024 +) + +// NewHbss2Pbss return a hash2Path obj +func NewHbss2Pbss(tr *Trie, db *Database, stateRootHash common.Hash, blocknum uint64, jobnum uint64) (*Hbss2Pbss, error) { + if tr == nil { + return nil, errors.New("trie is nil") + } + + if tr.root == nil { + return nil, errors.New("trie root is nil") + } + + ins := &Hbss2Pbss{ + trie: tr, + blocknum: blocknum, + db: db, + stateRootHash: stateRootHash, + root: tr.root, + concurrentQueue: make(chan struct{}, jobnum), + wg: sync.WaitGroup{}, + } + + return ins, nil +} + +func (t *Trie) resloveWithoutTrack(n node, prefix []byte) (node, error) { + if n, ok := n.(hashNode); ok { + blob, err := t.reader.node(prefix, common.BytesToHash(n)) + if err != nil { + return nil, err + } + return mustDecodeNode(n, blob), nil + } + return n, nil +} + +func (h2p *Hbss2Pbss) writeNode(pathKey []byte, n *trienode.Node, owner common.Hash) { + if owner == (common.Hash{}) { + rawdb.WriteAccountTrieNode(h2p.db.diskdb, pathKey, n.Blob) + log.Debug("WriteNodes account node, ", "path: ", common.Bytes2Hex(pathKey), "Hash: ", n.Hash, "BlobHash: ", crypto.Keccak256Hash(n.Blob)) + } else { + rawdb.WriteStorageTrieNode(h2p.db.diskdb, owner, pathKey, n.Blob) + log.Debug("WriteNodes storage node, ", "path: ", common.Bytes2Hex(pathKey), "owner: ", owner.String(), "Hash: ", n.Hash, "BlobHash: ", crypto.Keccak256Hash(n.Blob)) + } +} + +// Run statistics, external call +func (h2p *Hbss2Pbss) Run() { + log.Debug("Find Account Trie Tree, rootHash: ", h2p.trie.Hash().String(), "BlockNum: ", h2p.blocknum) + + h2p.ConcurrentTraversal(h2p.trie, h2p.root, []byte{}) + h2p.wg.Wait() + + log.Info("Total", "complete", h2p.totalNum, "go routines Num", runtime.NumGoroutine, "h2p concurrentQueue", len(h2p.concurrentQueue)) + + rawdb.WritePersistentStateID(h2p.db.diskdb, h2p.blocknum) + rawdb.WriteStateID(h2p.db.diskdb, h2p.stateRootHash, h2p.blocknum) +} + +func (h2p *Hbss2Pbss) SubConcurrentTraversal(theTrie *Trie, theNode node, path []byte) { + h2p.concurrentQueue <- struct{}{} + h2p.ConcurrentTraversal(theTrie, theNode, path) + <-h2p.concurrentQueue + h2p.wg.Done() +} + +func (h2p *Hbss2Pbss) ConcurrentTraversal(theTrie *Trie, theNode node, path []byte) { + total_num := uint64(0) + // nil node + if theNode == nil { + return + } + + switch current := (theNode).(type) { + case *shortNode: + collapsed := current.copy() + collapsed.Key = hexToCompact(current.Key) + var hash, _ = current.cache() + h2p.writeNode(path, trienode.New(common.BytesToHash(hash), nodeToBytes(collapsed)), theTrie.owner) + + h2p.ConcurrentTraversal(theTrie, current.Val, append(path, current.Key...)) + + case *fullNode: + // copy from trie/Committer (*committer).commit + collapsed := current.copy() + var hash, _ = collapsed.cache() + collapsed.Children = h2p.commitChildren(path, current) + + nodebytes := nodeToBytes(collapsed) + if common.BytesToHash(hash) != common.BytesToHash(crypto.Keccak256(nodebytes)) { + log.Error("Hash is inconsistent, hash: ", common.BytesToHash(hash), "node hash: ", common.BytesToHash(crypto.Keccak256(nodebytes)), "node: ", collapsed.fstring("")) + panic("hash inconsistent.") + } + + h2p.writeNode(path, trienode.New(common.BytesToHash(hash), nodeToBytes(collapsed)), theTrie.owner) + + for idx, child := range current.Children { + if child == nil { + continue + } + childPath := append(path, byte(idx)) + if len(h2p.concurrentQueue)*2 < cap(h2p.concurrentQueue) { + h2p.wg.Add(1) + dst := make([]byte, len(childPath)) + copy(dst, childPath) + go h2p.SubConcurrentTraversal(theTrie, child, dst) + } else { + h2p.ConcurrentTraversal(theTrie, child, childPath) + } + } + case hashNode: + n, err := theTrie.resloveWithoutTrack(current, path) + if err != nil { + log.Error("Resolve HashNode", "error", err, "TrieRoot", theTrie.Hash(), "Path", path) + return + } + h2p.ConcurrentTraversal(theTrie, n, path) + total_num = atomic.AddUint64(&h2p.totalNum, 1) + if total_num%100000 == 0 { + log.Info("Converting ", "Complete progress", total_num, "go routines Num", runtime.NumGoroutine(), "h2p concurrentQueue", len(h2p.concurrentQueue)) + } + return + case valueNode: + if !hasTerm(path) { + log.Info("ValueNode miss path term", "path", common.Bytes2Hex(path)) + break + } + var account types.StateAccount + if err := rlp.Decode(bytes.NewReader(current), &account); err != nil { + // log.Info("Rlp decode account failed.", "err", err) + break + } + if account.Root == (common.Hash{}) || account.Root == types.EmptyRootHash { + // log.Info("Not a storage trie.", "account", common.BytesToHash(path).String()) + break + } + + ownerAddress := common.BytesToHash(hexToCompact(path)) + tr, err := New(StorageTrieID(h2p.stateRootHash, ownerAddress, account.Root), h2p.db) + if err != nil { + log.Error("New Storage trie error", "err", err, "root", account.Root.String(), "owner", ownerAddress.String()) + break + } + log.Debug("Find Contract Trie Tree", "rootHash: ", tr.Hash().String(), "") + h2p.wg.Add(1) + go h2p.SubConcurrentTraversal(tr, tr.root, []byte{}) + default: + panic(errors.New("Invalid node type to traverse.")) + } +} + +// copy from trie/Commiter (*committer).commit +func (h2p *Hbss2Pbss) commitChildren(path []byte, n *fullNode) [17]node { + var children [17]node + for i := 0; i < 16; i++ { + child := n.Children[i] + if child == nil { + continue + } + // If it's the hashed child, save the hash value directly. + // Note: it's impossible that the child in range [0, 15] + // is a valueNode. + if hn, ok := child.(hashNode); ok { + children[i] = hn + continue + } + + children[i] = h2p.commit(append(path, byte(i)), child) + } + // For the 17th child, it's possible the type is valuenode. + if n.Children[16] != nil { + children[16] = n.Children[16] + } + return children +} + +// commit collapses a node down into a hash node and returns it. +func (h2p *Hbss2Pbss) commit(path []byte, n node) node { + // if this path is clean, use available cached data + hash, dirty := n.cache() + if hash != nil && !dirty { + return hash + } + // Commit children, then parent, and remove the dirty flag. + switch cn := n.(type) { + case *shortNode: + // Commit child + collapsed := cn.copy() + + // If the child is fullNode, recursively commit, + // otherwise it can only be hashNode or valueNode. + if _, ok := cn.Val.(*fullNode); ok { + collapsed.Val = h2p.commit(append(path, cn.Key...), cn.Val) + } + // The key needs to be copied, since we're adding it to the + // modified nodeset. + collapsed.Key = hexToCompact(cn.Key) + return collapsed + case *fullNode: + hashedKids := h2p.commitChildren(path, cn) + collapsed := cn.copy() + collapsed.Children = hashedKids + + return collapsed + case hashNode: + return cn + default: + // nil, valuenode shouldn't be committed + panic(fmt.Sprintf("%T: invalid node: %v", n, n)) + } +}