StateDiff over Sqlite3

cmd/geth/main.go

@@ -165,6 +165,7 @@ var (
 		// Enable verifier mode
 		utils.RollupEnableVerifierFlag,
 		utils.RollupAddressManagerOwnerAddressFlag,
+		utils.RollupDiffDbFlag,
 	}
 
 	rpcFlags = []cli.Flag{

cmd/geth/usage.go

@@ -78,6 +78,7 @@ var AppHelpFlagGroups = []flagGroup{
 			utils.Eth1HTTPFlag,
 			utils.RollupAddressManagerOwnerAddressFlag,
 			utils.RollupEnableVerifierFlag,
+			utils.RollupDiffDbFlag,
 		},
 	},
 	{

cmd/utils/flags.go

@@ -846,6 +846,12 @@ var (
 		Value:  "0x0000000000000000000000000000000000000000",
 		EnvVar: "ROLLUP_ADDRESS_MANAGER_OWNER_ADDRESS",
 	}
+	RollupDiffDbFlag = cli.Uint64Flag{
+		Name:   "rollup.diffdbcache",
+		Usage:  "Number of diffdb batch updates",
+		Value:  eth.DefaultConfig.DiffDbCache,
+		EnvVar: "ROLLUP_DIFFDB_CACHE",
+	}
 )
 
 // MakeDataDir retrieves the currently requested data directory, terminating
@@ -1603,6 +1609,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) {
 	setLes(ctx, cfg)
 	setEth1(ctx, &cfg.Rollup)
 
+	if ctx.GlobalIsSet(RollupDiffDbFlag.Name) {
+		cfg.DiffDbCache = ctx.GlobalUint64(RollupDiffDbFlag.Name)
+	}
 	if ctx.GlobalIsSet(SyncModeFlag.Name) {
 		cfg.SyncMode = *GlobalTextMarshaler(ctx, SyncModeFlag.Name).(*downloader.SyncMode)
 	}

core/blockchain.go

@@ -184,6 +184,20 @@ type BlockChain struct {
 	terminateInsert func(common.Hash, uint64) bool // Testing hook used to terminate ancient receipt chain insertion.
 }
 
+func NewBlockChainWithDiffDb(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(block *types.Block) bool, path string, cache uint64) (*BlockChain, error) {
+	diff, err := diffdb.NewDiffDb(path, cache)
+	if err != nil {
+		return nil, err
+	}
+	bc, err := NewBlockChain(db, cacheConfig, chainConfig, engine, vmConfig, shouldPreserve)
+	if err != nil {
+		return nil, err
+	}
+	bc.diffdb = diff
+
+	return bc, nil
+}
+
 // NewBlockChain returns a fully initialised block chain using information
 // available in the database. It initialises the default Ethereum Validator and
 // Processor.
@@ -203,16 +217,10 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
 	futureBlocks, _ := lru.New(maxFutureBlocks)
 	badBlocks, _ := lru.New(badBlockLimit)
 
-	diff, err := diffdb.NewDiffDb("eth/db/diffs")
-	if err != nil {
-		return nil, err
-	}
-
 	bc := &BlockChain{
 		chainConfig:    chainConfig,
 		cacheConfig:    cacheConfig,
 		db:             db,
-		diffdb:         diff,
 		triegc:         prque.New(nil),
 		stateCache:     state.NewDatabaseWithCache(db, cacheConfig.TrieCleanLimit),
 		quit:           make(chan struct{}),
@@ -231,6 +239,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
 	bc.prefetcher = newStatePrefetcher(chainConfig, bc, engine)
 	bc.processor = NewStateProcessor(chainConfig, bc, engine)
 
+	var err error
 	bc.hc, err = NewHeaderChain(db, chainConfig, engine, bc.getProcInterrupt)
 	if err != nil {
 		return nil, err
@@ -903,6 +912,15 @@ func (bc *BlockChain) Stop() {
 			log.Error("Dangling trie nodes after full cleanup")
 		}
 	}
+
+	if bc.diffdb != nil {
+		if err := bc.diffdb.ForceCommit(); err != nil {
+			log.Error("Failed to commit recent state diffs", "err", err)
+		}
+		if err := bc.diffdb.Close(); err != nil {
+			log.Error("Failed to commit state diffs handler", "err", err)
+		}
+	}
 	log.Info("Blockchain manager stopped")
 }
 

core/dao_test.go

@@ -122,7 +122,6 @@ func TestDAOForkRangeExtradata(t *testing.T) {
 	gspec.MustCommit(db)
 	bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil)
 	defer bc.Stop()
-
 	blocks := conBc.GetBlocksFromHash(conBc.CurrentBlock().Hash(), int(conBc.CurrentBlock().NumberU64()))
 	for j := 0; j < len(blocks)/2; j++ {
 		blocks[j], blocks[len(blocks)-1-j] = blocks[len(blocks)-1-j], blocks[j]

core/state/statedb.go

@@ -60,8 +60,10 @@ func (n *proofList) Delete(key []byte) error {
 
 // DiffDb is a database for storing state diffs per block
 type DiffDB interface {
-	SetDiffKey(*big.Int, common.Address, common.Hash, bool)
+	SetDiffKey(*big.Int, common.Address, common.Hash, bool) error
 	GetDiff(*big.Int) (diffdb.Diff, error)
+	Close() error
+	ForceCommit() error
 }
 
 // StateDBs within the ethereum protocol are used to store anything
@@ -150,8 +152,7 @@ func (s *StateDB) SetDiffKey(block *big.Int, address common.Address, key common.
 	if s.diffdb == nil {
 		return errors.New("DiffDB not set")
 	}
-	s.diffdb.SetDiffKey(block, address, key, mutated)
-	return nil
+	return s.diffdb.SetDiffKey(block, address, key, mutated)
 }
 
 func (s *StateDB) Error() error {

core/vm/ovm_state_manager_test.go

@@ -3,6 +3,7 @@ package vm
 import (
 	"crypto/rand"
 	"math/big"
+	"os"
 	"reflect"
 	"testing"
 
@@ -28,23 +29,21 @@ type ContractData struct {
 var (
 	contract1 = common.HexToAddress("0x000000000000000000000000000000000001")
 	contract2 = common.HexToAddress("0x000000000000000000000000000000000002")
-	env       *EVM
-	contract  *Contract
-	db        *diffdb.DiffDb
-	mock      *mockDb
-	testData  TestData
 )
 
-func init() {
-	db, _ = diffdb.NewDiffDb("test")
-	mock = &mockDb{db: *db}
-	env = NewEVM(Context{}, mock, params.TestChainConfig, Config{})
+func makeEnv(dbname string) (*diffdb.DiffDb, *EVM, TestData, *Contract) {
+	db, _ := diffdb.NewDiffDb(dbname, 1)
+	mock := &mockDb{db: *db}
+	env := NewEVM(Context{}, mock, params.TestChainConfig, Config{})
 	// re-use `dummyContractRef` from `logger_test.go`
-	contract = NewContract(&dummyContractRef{}, &dummyContractRef{}, new(big.Int), 0)
-	testData = make(TestData)
+	contract := NewContract(&dummyContractRef{}, &dummyContractRef{}, new(big.Int), 0)
+	testData := make(TestData)
+	return db, env, testData, contract
 }
 
 func TestEthCallNoop(t *testing.T) {
+	db, env, _, contract := makeEnv("test1")
+	defer os.Remove("test1")
 	env.Context.EthCallSender = &common.Address{0}
 	env.Context.BlockNumber = big.NewInt(1)
 	args := map[string]interface{}{
@@ -53,13 +52,18 @@ func TestEthCallNoop(t *testing.T) {
 		"_value":    [32]uint8{2},
 	}
 	putContractStorage(env, contract, args)
-	diff, _ := db.GetDiff(env.Context.BlockNumber)
+	diff, err := db.GetDiff(env.Context.BlockNumber)
+	if err != nil {
+		t.Fatal("Db call error", err)
+	}
 	if len(diff) > 0 {
 		t.Fatalf("map must be empty since it was an eth call")
 	}
 }
 
 func TestSetDiffs(t *testing.T) {
+	db, env, testData, contract := makeEnv("test2")
+	defer os.Remove("test2")
 	// not an eth-call
 	env.Context.EthCallSender = nil
 	// in block 1 both contracts get touched
@@ -82,7 +86,10 @@ func TestSetDiffs(t *testing.T) {
 	}
 
 	// empty diff for the next block
-	diff2, _ := db.GetDiff(blockNumber2)
+	diff2, err := db.GetDiff(blockNumber2)
+	if err != nil {
+		t.Fatal("Db call error", err)
+	}
 	if len(diff2) != 0 {
 		t.Fatalf("Diff2 should be empty since data about the next block is not added yet")
 	}
@@ -91,7 +98,10 @@ func TestSetDiffs(t *testing.T) {
 	putTestData(t, env, contract, blockNumber2, testData)
 
 	expected2 := getExpected(testData[blockNumber2])
-	diff2, _ = db.GetDiff(blockNumber2)
+	diff2, err = db.GetDiff(blockNumber2)
+	if err != nil {
+		t.Fatal("Db call error", err)
+	}
 	if !reflect.DeepEqual(diff2, expected2) {
 		t.Fatalf("Diff2 did not match.")
 	}

diffdb/db.go

@@ -1,10 +1,11 @@
 package diffdb
 
 import (
-	"errors"
-	"math/big"
-
 	"github.com/ethereum/go-ethereum/common"
+	_ "github.com/mattn/go-sqlite3"
+
+	"database/sql"
+	"math/big"
 )
 
 type Key struct {
@@ -14,38 +15,140 @@ type Key struct {
 
 type Diff map[common.Address][]Key
 
+/// A DiffDb is a thin wrapper around an Sqlite3 connection.
+///
+/// Its purpose is to store and fetch the storage keys corresponding to an address that was
+/// touched in a block.
 type DiffDb struct {
-	// Todo: should this be go-ethereum's leveldb maybe?
-	// db *leveldb.DB
-	inner map[uint64]Diff
+	db    *sql.DB
+	tx    *sql.Tx
+	stmt  *sql.Stmt
+	cache uint64
+	// We have a db-wide counter for the number of db calls made which we reset
+	// whenever it hits `cache`.
+	numCalls uint64
+}
+
+var insertStatement = `
+INSERT INTO diffs
+    (block, address, key, mutated)
+    VALUES
+    ($1, $2, $3, $4)
+`
+var createStmt = `
+CREATE TABLE IF NOT EXISTS diffs (
+    block INTEGER,
+    address STRING,
+    key STRING,
+    mutated BOOL
+)
+`
+var selectStmt = `
+SELECT * from diffs WHERE block = $1
+`
+
+/// Inserts a new row to the sqlite with the provided diff data.
+func (diff *DiffDb) SetDiffKey(block *big.Int, address common.Address, key common.Hash, mutated bool) error {
+	// add 1 more insertion to the transaction
+	_, err := diff.stmt.Exec(block.Uint64(), address, key, mutated)
+	if err != nil {
+		return err
+	}
+
+	// increment number of calls
+	diff.numCalls += 1
+
+	// if we had enough calls, commit it
+	if diff.numCalls == diff.cache {
+		if err := diff.ForceCommit(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+/// Commits a pending diffdb transaction
+func (diff *DiffDb) ForceCommit() error {
+	if err := diff.tx.Commit(); err != nil {
+		return err
+	}
+	return diff.resetTx()
 }
 
-// Called by the OVM StateManager
-func (diff DiffDb) SetDiffKey(block *big.Int, address common.Address, key common.Hash, mutated bool) {
-	// instantiate the diff
-	if diff.inner[block.Uint64()] == nil {
-		diff.inner[block.Uint64()] = make(map[common.Address][]Key)
+/// Gets all the rows for the matching block and converts them to a Diff map.
+func (diff *DiffDb) GetDiff(blockNum *big.Int) (Diff, error) {
+	// make the query
+	rows, err := diff.db.Query(selectStmt, blockNum.Uint64())
+	if err != nil {
+		return nil, err
+	}
+
+	// initialize our data
+	res := make(Diff)
+	var block uint64
+	var address common.Address
+	var key common.Hash
+	var mutated bool
+	for rows.Next() {
+		// deserialize the line
+		err = rows.Scan(&block, &address, &key, &mutated)
+		if err != nil {
+			return nil, err
+		}
+		// add the data to the map
+		res[address] = append(res[address], Key{key, mutated})
 	}
 
-	// set the value
-	diff.inner[block.Uint64()][address] = append(diff.inner[block.Uint64()][address], Key{key, mutated})
+	return res, rows.Err()
 }
 
-/// Gets a list of diffs from the databse for the corresponding
-func (diff *DiffDb) GetDiff(block *big.Int) (Diff, error) {
-	res, ok := diff.inner[block.Uint64()]
-	if !ok {
-		return nil, errors.New("No diff was found for the provided block")
+// Initializes the transaction which we will be using to commit data to the db
+func (diff *DiffDb) resetTx() error {
+	// reset the number of calls made
+	diff.numCalls = 0
+
+	// start a new tx
+	tx, err := diff.db.Begin()
+	if err != nil {
+		return err
 	}
-	return res, nil
+	diff.tx = tx
+
+	// the tx is about inserts
+	stmt, err := diff.tx.Prepare(insertStatement)
+	if err != nil {
+		return err
+	}
+	diff.stmt = stmt
+
+	return nil
 }
 
-func NewDiffDb(path string) (*DiffDb, error) {
-	// db, err := leveldb.OpenFile(path, nil)
-	// if err != nil {
-	//     return nil, err
-	// }
-	// return &DiffDb{ db: db }, nil
-	diffdb := make(map[uint64]Diff)
-	return &DiffDb{inner: diffdb}, nil
+func (diff *DiffDb) Close() error {
+	return diff.db.Close()
+}
+
+/// Instantiates a new DiffDb using sqlite at `path`, with `cache` insertions
+/// done in a transaction before it gets committed to the database.
+func NewDiffDb(path string, cache uint64) (*DiffDb, error) {
+	// get a handle
+	db, err := sql.Open("sqlite3", path)
+	if err != nil {
+		return nil, err
+	}
+
+	// create the table if it does not exist
+	_, err = db.Exec(createStmt)
+	if err != nil {
+		return nil, err
+	}
+
+	diffdb := &DiffDb{db: db, cache: cache}
+
+	// initialize the transaction
+	if err := diffdb.resetTx(); err != nil {
+		return nil, err
+	}
+	return diffdb, nil
 }