core, ethdb, trie: experiment with batch-replay-based uncaching

core/rawdb/table.go

@@ -124,6 +124,10 @@ type tableBatch struct {
 	prefix string
 }
 
+func (b *tableBatch) Replay(replay ethdb.DbEventLogger) error {
+	panic("implement me")
+}
+
 // Put inserts the given value into the batch for later committing.
 func (b *tableBatch) Put(key, value []byte) error {
 	return b.batch.Put(append([]byte(b.prefix), key...), value)

ethdb/batch.go

@@ -34,6 +34,9 @@ type Batch interface {
 
 	// Reset resets the batch for reuse
 	Reset()
+
+	// Replay replays the batch into another batch
+	Replay(logger DbEventLogger) error
 }
 
 // Batcher wraps the NewBatch method of a backing data store.

ethdb/database.go

@@ -40,6 +40,13 @@ type Deleter interface {
 	Delete(key []byte) error
 }
 
+// DbEventLogger wraps Put and Delete to serve as a recipient
+// for batch replays
+type DbEventLogger interface {
+	Writer
+	Deleter
+}
+
 // Stater wraps the Stat method of a backing data store.
 type Stater interface {
 	// Stat returns a particular internal stat of the database.

ethdb/leveldb/leveldb.go

@@ -172,6 +172,23 @@ func (db *Database) NewBatch() ethdb.Batch {
 	}
 }
 
+type dbWrapper struct {
+	wrapped ethdb.DbEventLogger
+}
+
+func (dbw *dbWrapper) Put(key, value []byte) {
+	dbw.wrapped.Put(key, value)
+}
+
+func (dbw *dbWrapper) Delete(key []byte) {
+	dbw.wrapped.Delete(key)
+}
+
+// Replay replays batch contents.
+func (b *batch) Replay(r ethdb.DbEventLogger) error {
+	return b.b.Replay(&dbWrapper{r})
+}
+
 // NewIterator creates a binary-alphabetical iterator over the entire keyspace
 // contained within the leveldb database.
 func (db *Database) NewIterator() ethdb.Iterator {

ethdb/memorydb/memorydb.go

@@ -200,6 +200,17 @@ type batch struct {
 	size   int
 }
 
+func (b *batch) Replay(replay ethdb.DbEventLogger) error {
+	for _, keyvalue := range b.writes {
+		if keyvalue.delete {
+			replay.Delete(keyvalue.key)
+			continue
+		}
+		replay.Put(keyvalue.key, keyvalue.value)
+	}
+	return nil
+}
+
 // Put inserts the given value into the batch for later committing.
 func (b *batch) Put(key, value []byte) error {
 	b.writes = append(b.writes, keyvalue{common.CopyBytes(key), common.CopyBytes(value), false})

trie/database.go

@@ -81,7 +81,8 @@ type Database struct {
 	dirtiesSize   common.StorageSize // Storage size of the dirty node cache (exc. flushlist)
 	preimagesSize common.StorageSize // Storage size of the preimages cache
 
-	lock sync.RWMutex
+	lock        sync.RWMutex
+	batchLogger *BatchEventLogger
 }
 
 // rawNode is a simple binary blob used to differentiate between collapsed trie
@@ -293,12 +294,14 @@ func NewDatabaseWithCache(diskdb ethdb.KeyValueStore, cache int) *Database {
 			Hasher:             trienodeHasher{},
 		})
 	}
-	return &Database{
+	db := &Database{
 		diskdb:    diskdb,
 		cleans:    cleans,
 		dirties:   map[common.Hash]*cachedNode{{}: {}},
 		preimages: make(map[common.Hash][]byte),
 	}
+	db.batchLogger = newBatchEventLogger(db)
+	return db
 }
 
 // DiskDB retrieves the persistent storage backing the trie database.
@@ -660,6 +663,61 @@ func (db *Database) Cap(limit common.StorageSize) error {
 	return nil
 }
 
+type BatchEventLogger struct {
+	db *Database
+}
+
+func newBatchEventLogger(db *Database) *BatchEventLogger {
+	return &BatchEventLogger{db}
+}
+
+// Put reacts to batch writes, and implements uncache:
+// is the post-processing step of a commit operation where the already
+// persisted trie is removed from the cache. The reason behind the two-phase
+// commit is to ensure consistent data availability while moving from memory
+// to disk.
+func (p *BatchEventLogger) Put(key []byte, value []byte) error {
+	// key is hash
+	// value is rlp
+	//log.Info("proxybatch", "key", fmt.Sprintf("0x%x", key))
+	hash := common.BytesToHash(key)
+	db := p.db
+	rlp := value
+	// If the node does not exist, we're done on this path
+	node, ok := db.dirties[hash]
+	if !ok {
+		return nil
+	}
+	// Node still exists, remove it from the flush-list
+	switch hash {
+	case db.oldest:
+		db.oldest = node.flushNext
+		db.dirties[node.flushNext].flushPrev = common.Hash{}
+	case db.newest:
+		db.newest = node.flushPrev
+		db.dirties[node.flushPrev].flushNext = common.Hash{}
+	default:
+		db.dirties[node.flushPrev].flushNext = node.flushNext
+		db.dirties[node.flushNext].flushPrev = node.flushPrev
+	}
+	// Uncache the node's subtries and remove the node itself too
+	//for _, child := range node.childs() {
+	//	db.uncache(child)
+	//}
+	delete(db.dirties, hash)
+	db.dirtiesSize -= common.StorageSize(common.HashLength + int(node.size))
+
+	// Move the flushed node into the clean cache to prevent insta-reloads
+	if db.cleans != nil {
+		db.cleans.Set(string(hash[:]), rlp)
+	}
+	return nil
+}
+
+func (p *BatchEventLogger) Delete(key []byte) error {
+	panic("Not implemented")
+}
+
 // Commit iterates over all the children of a particular node, writes them out
 // to disk, forcefully tearing down all references in both directions.
 //
@@ -673,7 +731,6 @@ func (db *Database) Commit(node common.Hash, report bool) error {
 
 	start := time.Now()
 	batch := db.diskdb.NewBatch()
-
 	// Move all of the accumulated preimages into a write batch
 	for hash, preimage := range db.preimages {
 		if err := batch.Put(db.secureKey(hash[:]), preimage); err != nil {
@@ -689,6 +746,11 @@ func (db *Database) Commit(node common.Hash, report bool) error {
 			batch.Reset()
 		}
 	}
+	if err := batch.Write(); err != nil {
+		db.lock.RUnlock()
+		return err
+	}
+	batch.Reset()
 	// Move the trie itself into the batch, flushing if enough data is accumulated
 	nodes, storage := len(db.dirties), db.dirtiesSize
 	if err := db.commit(node, batch); err != nil {
@@ -703,16 +765,15 @@ func (db *Database) Commit(node common.Hash, report bool) error {
 		return err
 	}
 	db.lock.RUnlock()
-
 	// Write successful, clear out the flushed data
 	db.lock.Lock()
 	defer db.lock.Unlock()
+	batch.Replay(db.batchLogger)
+	batch.Reset()
 
 	db.preimages = make(map[common.Hash][]byte)
 	db.preimagesSize = 0
 
-	db.uncache(node)
-
 	memcacheCommitTimeTimer.Update(time.Since(start))
 	memcacheCommitSizeMeter.Mark(int64(storage - db.dirtiesSize))
 	memcacheCommitNodesMeter.Mark(int64(nodes - len(db.dirties)))
@@ -751,46 +812,18 @@ func (db *Database) commit(hash common.Hash, batch ethdb.Batch) error {
 		if err := batch.Write(); err != nil {
 			return err
 		}
-		batch.Reset()
+		db.lock.RUnlock()
+		{
+			db.lock.Lock()
+			batch.Replay(db.batchLogger)
+			batch.Reset()
+			db.lock.Unlock()
+		}
+		db.lock.RLock()
 	}
 	return nil
 }
 
-// uncache is the post-processing step of a commit operation where the already
-// persisted trie is removed from the cache. The reason behind the two-phase
-// commit is to ensure consistent data availability while moving from memory
-// to disk.
-func (db *Database) uncache(hash common.Hash) {
-	// If the node does not exist, we're done on this path
-	node, ok := db.dirties[hash]
-	if !ok {
-		return
-	}
-	// Node still exists, remove it from the flush-list
-	switch hash {
-	case db.oldest:
-		db.oldest = node.flushNext
-		db.dirties[node.flushNext].flushPrev = common.Hash{}
-	case db.newest:
-		db.newest = node.flushPrev
-		db.dirties[node.flushPrev].flushNext = common.Hash{}
-	default:
-		db.dirties[node.flushPrev].flushNext = node.flushNext
-		db.dirties[node.flushNext].flushPrev = node.flushPrev
-	}
-	// Uncache the node's subtries and remove the node itself too
-	for _, child := range node.childs() {
-		db.uncache(child)
-	}
-	delete(db.dirties, hash)
-	db.dirtiesSize -= common.StorageSize(common.HashLength + int(node.size))
-
-	// Move the flushed node into the clean cache to prevent insta-reloads
-	if db.cleans != nil {
-		db.cleans.Set(string(hash[:]), node.rlp())
-	}
-}
-
 // Size returns the current storage size of the memory cache in front of the
 // persistent database layer.
 func (db *Database) Size() (common.StorageSize, common.StorageSize) {