From 887e7164306b5f39a5a3aa884fc2cc9a67d2b5c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 25 Mar 2019 13:20:17 +0200 Subject: [PATCH] core, ethdb, trie: polish up batch replay mechanism --- core/rawdb/table.go | 9 +-- ethdb/batch.go | 6 +- ethdb/database.go | 8 +-- ethdb/leveldb/leveldb.go | 46 ++++++++----- ethdb/memorydb/memorydb.go | 23 +++---- trie/database.go | 131 ++++++++++++++++++------------------- 6 files changed, 118 insertions(+), 105 deletions(-) diff --git a/core/rawdb/table.go b/core/rawdb/table.go index 7ea9b2c7db..04f81f8ee0 100644 --- a/core/rawdb/table.go +++ b/core/rawdb/table.go @@ -124,10 +124,6 @@ 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) @@ -152,3 +148,8 @@ func (b *tableBatch) Write() error { func (b *tableBatch) Reset() { b.batch.Reset() } + +// Replay replays the batch contents. +func (b *tableBatch) Replay(r ethdb.Replayee) error { + return b.batch.Replay(r) +} diff --git a/ethdb/batch.go b/ethdb/batch.go index 38b704b0eb..538b464ccd 100644 --- a/ethdb/batch.go +++ b/ethdb/batch.go @@ -32,11 +32,11 @@ type Batch interface { // Write flushes any accumulated data to disk. Write() error - // Reset resets the batch for reuse + // Reset resets the batch for reuse. Reset() - // Replay replays the batch into another batch - Replay(logger DbEventLogger) error + // Replay replays the batch contents. + Replay(replayer Replayee) error } // Batcher wraps the NewBatch method of a backing data store. diff --git a/ethdb/database.go b/ethdb/database.go index 35ce8a4d25..ddb1fd06e9 100644 --- a/ethdb/database.go +++ b/ethdb/database.go @@ -14,7 +14,7 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -// Package database defines the interfaces for an Ethereum data store. +// Package ethdb defines the interfaces for an Ethereum data store. package ethdb import "io" @@ -40,9 +40,9 @@ type Deleter interface { Delete(key []byte) error } -// DbEventLogger wraps Put and Delete to serve as a recipient -// for batch replays -type DbEventLogger interface { +// Replayee wraps basic batch operations to allow replaying an existing batch +// on top of multiple databases. +type Replayee interface { Writer Deleter } diff --git a/ethdb/leveldb/leveldb.go b/ethdb/leveldb/leveldb.go index 835fab4828..c11b39bfbe 100644 --- a/ethdb/leveldb/leveldb.go +++ b/ethdb/leveldb/leveldb.go @@ -172,23 +172,6 @@ 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 { @@ -433,3 +416,32 @@ func (b *batch) Reset() { b.b.Reset() b.size = 0 } + +// Replay replays the batch contents. +func (b *batch) Replay(r ethdb.Replayee) error { + return b.b.Replay(&replayer{replayer: r}) +} + +// replayer is a small wrapper to implement the correct replay methods. +type replayer struct { + replayer ethdb.Replayee + failure error +} + +// Put inserts the given value into the key-value data store. +func (r *replayer) Put(key, value []byte) { + // If the replay already failed, stop executing ops + if r.failure != nil { + return + } + r.failure = r.replayer.Put(key, value) +} + +// Delete removes the key from the key-value data store. +func (r *replayer) Delete(key []byte) { + // If the replay already failed, stop executing ops + if r.failure != nil { + return + } + r.failure = r.replayer.Delete(key) +} diff --git a/ethdb/memorydb/memorydb.go b/ethdb/memorydb/memorydb.go index db591a7276..55d88d2135 100644 --- a/ethdb/memorydb/memorydb.go +++ b/ethdb/memorydb/memorydb.go @@ -200,17 +200,6 @@ 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}) @@ -251,6 +240,18 @@ func (b *batch) Reset() { b.size = 0 } +// Replay replays the batch contents. +func (b *batch) Replay(r ethdb.Replayee) error { + for _, keyvalue := range b.writes { + if keyvalue.delete { + r.Delete(keyvalue.key) + continue + } + r.Put(keyvalue.key, keyvalue.value) + } + return nil +} + // iterator can walk over the (potentially partial) keyspace of a memory key // value store. Internally it is a deep copy of the entire iterated state, // sorted by keys. diff --git a/trie/database.go b/trie/database.go index 7193b3324f..620b13b56d 100644 --- a/trie/database.go +++ b/trie/database.go @@ -81,8 +81,7 @@ 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 - batchLogger *BatchEventLogger + lock sync.RWMutex } // rawNode is a simple binary blob used to differentiate between collapsed trie @@ -300,7 +299,6 @@ func NewDatabaseWithCache(diskdb ethdb.KeyValueStore, cache int) *Database { dirties: map[common.Hash]*cachedNode{{}: {}}, preimages: make(map[common.Hash][]byte), } - db.batchLogger = newBatchEventLogger(db) return db } @@ -663,61 +661,6 @@ 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. // @@ -731,6 +674,7 @@ 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 { @@ -738,6 +682,7 @@ func (db *Database) Commit(node common.Hash, report bool) error { db.lock.RUnlock() return err } + // If the batch is too large, flush to disk if batch.ValueSize() > ethdb.IdealBatchSize { if err := batch.Write(); err != nil { db.lock.RUnlock() @@ -746,31 +691,39 @@ func (db *Database) Commit(node common.Hash, report bool) error { batch.Reset() } } + // Since we're going to replay trie node writes into the clean cache, flush out + // any batched pre-images before continuing. 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 { + + uncacher := &cleaner{db} + if err := db.commit(node, batch, uncacher); err != nil { log.Error("Failed to commit trie from trie database", "err", err) db.lock.RUnlock() return err } - // Write batch ready, unlock for readers during persistence + // Trie mostly committed to disk, flush any batch leftovers if err := batch.Write(); err != nil { log.Error("Failed to write trie to disk", "err", err) db.lock.RUnlock() return err } db.lock.RUnlock() - // Write successful, clear out the flushed data + + // Uncache any leftovers in the last batch db.lock.Lock() defer db.lock.Unlock() - batch.Replay(db.batchLogger) + + batch.Replay(uncacher) batch.Reset() + // Reset the storage counters and bumpd metrics db.preimages = make(map[common.Hash][]byte) db.preimagesSize = 0 @@ -793,14 +746,14 @@ func (db *Database) Commit(node common.Hash, report bool) error { } // commit is the private locked version of Commit. -func (db *Database) commit(hash common.Hash, batch ethdb.Batch) error { +func (db *Database) commit(hash common.Hash, batch ethdb.Batch, uncacher *cleaner) error { // If the node does not exist, it's a previously committed node node, ok := db.dirties[hash] if !ok { return nil } for _, child := range node.childs() { - if err := db.commit(child, batch); err != nil { + if err := db.commit(child, batch, uncacher); err != nil { return err } } @@ -815,15 +768,61 @@ func (db *Database) commit(hash common.Hash, batch ethdb.Batch) error { db.lock.RUnlock() { db.lock.Lock() - batch.Replay(db.batchLogger) - batch.Reset() + batch.Replay(uncacher) db.lock.Unlock() + batch.Reset() } db.lock.RLock() } return nil } +// cleaner is a database batch replayer that takes a batch of write operations +// and cleans up the trie database from anything written to disk. +type cleaner struct { + db *Database +} + +// Put reacts to database writes and implements dirty data uncaching. This is the +// post-processing step of a commit operation where the already persisted trie is +// removed from the dirty cache and moved into the clean cache. The reason behind +// the two-phase commit is to ensure ensure data availability while moving from +// memory to disk. +func (c *cleaner) Put(key []byte, rlp []byte) error { + hash := common.BytesToHash(key) + + // If the node does not exist, we're done on this path + node, ok := c.db.dirties[hash] + if !ok { + return nil + } + // Node still exists, remove it from the flush-list + switch hash { + case c.db.oldest: + c.db.oldest = node.flushNext + c.db.dirties[node.flushNext].flushPrev = common.Hash{} + case c.db.newest: + c.db.newest = node.flushPrev + c.db.dirties[node.flushPrev].flushNext = common.Hash{} + default: + c.db.dirties[node.flushPrev].flushNext = node.flushNext + c.db.dirties[node.flushNext].flushPrev = node.flushPrev + } + // Remove the node from the dirty cache + delete(c.db.dirties, hash) + c.db.dirtiesSize -= common.StorageSize(common.HashLength + int(node.size)) + + // Move the flushed node into the clean cache to prevent insta-reloads + if c.db.cleans != nil { + c.db.cleans.Set(string(hash[:]), rlp) + } + return nil +} + +func (c *cleaner) Delete(key []byte) error { + panic("Not implemented") +} + // 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) {