ethdb, cmd/geth, eth: guerilla state pruning

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.")

cmd/geth/main.go

@@ -95,6 +95,7 @@ func init() {
 		monitorCommand,
 		accountCommand,
 		walletCommand,
+		pruningCommand,
 		{
 			Action: makedag,
 			Name:   "makedag",

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

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)
 }

ethdb/interface.go

@@ -26,5 +26,6 @@ type Database interface {
 
 type Batch interface {
 	Put(key, value []byte) error
+	Delete(key []byte) error
 	Write() error
 }

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
 }