diff --git a/core/rawdb/table.go b/core/rawdb/table.go index 974df681bd..7ea9b2c7db 100644 --- a/core/rawdb/table.go +++ b/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) diff --git a/ethdb/batch.go b/ethdb/batch.go index a6f0158210..38b704b0eb 100644 --- a/ethdb/batch.go +++ b/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. diff --git a/ethdb/database.go b/ethdb/database.go index 30208e1468..35ce8a4d25 100644 --- a/ethdb/database.go +++ b/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. diff --git a/ethdb/leveldb/leveldb.go b/ethdb/leveldb/leveldb.go index f730887d25..835fab4828 100644 --- a/ethdb/leveldb/leveldb.go +++ b/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 { diff --git a/ethdb/memorydb/memorydb.go b/ethdb/memorydb/memorydb.go index cc2490ba2f..db591a7276 100644 --- a/ethdb/memorydb/memorydb.go +++ b/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}) diff --git a/trie/database.go b/trie/database.go index a7106b60db..7193b3324f 100644 --- a/trie/database.go +++ b/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) {