From 0952a44ebaf41cb9591e8bb230f11e782f8ba016 Mon Sep 17 00:00:00 2001 From: Jeffrey Wilcke Date: Mon, 25 Apr 2016 11:04:11 +0200 Subject: [PATCH] ethdb, cmd/geth, eth: guerilla state pruning --- cmd/geth/chaincmd.go | 133 +++++++++++++++++++++++++++++++++++++++ cmd/geth/main.go | 1 + cmd/utils/flags.go | 14 +++++ ethdb/database.go | 5 ++ ethdb/interface.go | 1 + ethdb/memory_database.go | 18 +++++- 6 files changed, 169 insertions(+), 3 deletions(-) diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 32eacc99ee..2d374945c8 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -30,7 +30,10 @@ import ( "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/trie" + "github.com/syndtr/goleveldb/leveldb/util" ) var ( @@ -69,8 +72,138 @@ The arguments are interpreted as block numbers or hashes. Use "ethereum dump 0" to dump the genesis block. `, } + pruningCommand = cli.Command{ + Action: pruneDB, + Name: "prune", + Usage: "Prunes database of old state information", + } ) +func pruneDB(ctx *cli.Context) { + chainDb := utils.MakeChainDatabase(ctx) + defer chainDb.Close() + + // create a new temporary database to which we'll copy + // the required state information. This DB will be + // removed once finished. + const pruningDB = "data_pruning_process" + fresh := utils.MustOpenDatabase(ctx, pruningDB) + defer func() { + fresh.Close() + os.RemoveAll(filepath.Join(utils.MustMakeDataDir(ctx), pruningDB)) + }() + + /* + if db, ok := chainDb.(*ethdb.LDBDatabase); ok { + iter := db.NewIterator() + for iter.Next() { + key := iter.Key() + fmt.Printf("(%-4d) %-6d %x\n", len(key), len(iter.Value()), key) + //db.Delete(key) + } + iter.Release() + } + return + */ + + tbegin := time.Now() + glog.V(logger.Info).Infoln("Starting pruning process (this may take a while)") + + // Fetch the current head block on which we'll determine the + // "required state information" (current - 256 blocks ago) + headHash := core.GetHeadBlockHash(chainDb) + if (headHash == common.Hash{}) { + utils.Fatalf("pruning: no HEAD block found") + } + headBlock := core.GetBlock(chainDb, headHash) + + glog.V(logger.Info).Infoln("Copying relevant state data") + + // Fetch each block and start the copying process. The copying process will use the + // state iterator to check whether it needs to copy over state nodes, if a node is + // missing it will fetch it from the database and puts it in our new fresh database. + for blockno := headBlock.NumberU64() - 1500; blockno <= headBlock.NumberU64(); blockno++ { + numhash := 0 + + // fetch the block for the state root, this is the start of + // the copying process + hash := core.GetCanonicalHash(chainDb, blockno) + if (hash == common.Hash{}) { + utils.Fatalf("pruning: unable to find block %d for pruning session", blockno) + } + stateRoot := core.GetBlock(chainDb, hash).Root() + + // create a new sync state for the copying process + sync := state.NewStateSync(stateRoot, fresh) + // find missing nodes in batches of 256 and copy over the data. + for missing := sync.Missing(256); len(missing) > 0; missing = sync.Missing(256) { + syncRes := make([]trie.SyncResult, len(missing)) + for i, hash := range missing { + node, _ := chainDb.Get(hash[:]) + + syncRes[i] = trie.SyncResult{Hash: hash, Data: node} + } + sync.Process(syncRes) + numhash += len(syncRes) + } + } + + glog.V(logger.Info).Infoln("Pruning old state data") + + var ( + tmp = fresh.(*ethdb.LDBDatabase) + chain = chainDb.(*ethdb.LDBDatabase) + ) + // Deletion process. The deletion process will fetch the secure-keys + // and deletes them from the database. + presec := []byte("secure-key-") + // Unfortunately our current implementation of the DB wrapper + // does not allow us to create iterators whith filtering + // options and therefor we fetch the leveldb instance and + // create a iterator from there instead. + { + iter := chain.LDB().NewIterator(util.BytesPrefix(presec), nil) + for iter.Next() { + // delete each key + chain.Delete(iter.Key()) + } + iter.Release() + } + + // delete old state information. this process is not yet finished + // this process curerntly checks whether the key is 32 and **assumes** + // it's state information. + { + iter := chain.NewIterator() + for iter.Next() { + key := iter.Key() + if len(key) == 32 { + chain.Delete(key) + } + } + iter.Release() + } + + glog.V(logger.Info).Infoln("Compacting database") + // Compact the database again + chain.LDB().CompactRange(util.Range{nil, nil}) + + glog.V(logger.Info).Infoln("Moving new state data to database") + + // Copy the relevant state information back to the state db + { + batch := chain.NewBatch() + iter := tmp.NewIterator() + for iter.Next() { + batch.Put(iter.Key(), iter.Value()) + } + iter.Release() + batch.Write() + } + + glog.V(logger.Info).Infoln("Pruning process completed in:", time.Since(tbegin)) +} + func importChain(ctx *cli.Context) { if len(ctx.Args()) != 1 { utils.Fatalf("This command requires an argument.") diff --git a/cmd/geth/main.go b/cmd/geth/main.go index b8e2a78b81..3f0a374c8d 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -95,6 +95,7 @@ func init() { monitorCommand, accountCommand, walletCommand, + pruningCommand, { Action: makedag, Name: "makedag", diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 43dbc37f74..c1623f344b 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -841,6 +841,20 @@ func MakeChainDatabase(ctx *cli.Context) ethdb.Database { return chainDb } +func MustOpenDatabase(ctx *cli.Context, name string) ethdb.Database { + var ( + datadir = MustMakeDataDir(ctx) + cache = ctx.GlobalInt(CacheFlag.Name) + handles = MakeDatabaseHandles() + ) + + db, err := ethdb.NewLDBDatabase(filepath.Join(datadir, name), cache, handles) + if err != nil { + Fatalf("Could not open database '%s': %v", name, err) + } + return db +} + // MakeChain creates a chain manager from set command line flags. func MakeChain(ctx *cli.Context) (chain *core.BlockChain, chainDb ethdb.Database) { var err error diff --git a/ethdb/database.go b/ethdb/database.go index dffb42e2b0..4291084f65 100644 --- a/ethdb/database.go +++ b/ethdb/database.go @@ -295,6 +295,11 @@ func (b *ldbBatch) Put(key, value []byte) error { return nil } +func (b *ldbBatch) Delete(key []byte) error { + b.b.Delete(key) + return nil +} + func (b *ldbBatch) Write() error { return b.db.Write(b.b, nil) } diff --git a/ethdb/interface.go b/ethdb/interface.go index f4b787a52a..e394b1924d 100644 --- a/ethdb/interface.go +++ b/ethdb/interface.go @@ -26,5 +26,6 @@ type Database interface { type Batch interface { Put(key, value []byte) error + Delete(key []byte) error Write() error } diff --git a/ethdb/memory_database.go b/ethdb/memory_database.go index a729f52332..122865eff2 100644 --- a/ethdb/memory_database.go +++ b/ethdb/memory_database.go @@ -98,9 +98,10 @@ func (db *MemDatabase) NewBatch() Batch { type kv struct{ k, v []byte } type memBatch struct { - db *MemDatabase - writes []kv - lock sync.RWMutex + db *MemDatabase + writes []kv + deletes []kv + lock sync.RWMutex } func (b *memBatch) Put(key, value []byte) error { @@ -111,6 +112,14 @@ func (b *memBatch) Put(key, value []byte) error { return nil } +func (b *memBatch) Delete(key []byte) error { + b.lock.Lock() + defer b.lock.Unlock() + + b.deletes = append(b.deletes, kv{common.CopyBytes(key), nil}) + return nil +} + func (b *memBatch) Write() error { b.lock.RLock() defer b.lock.RUnlock() @@ -121,5 +130,8 @@ func (b *memBatch) Write() error { for _, kv := range b.writes { b.db.db[string(kv.k)] = kv.v } + for _, kv := range b.deletes { + delete(b.db.db, string(kv.k)) + } return nil }