package ffldb import ( "bytes" "fmt" "sync" "time" "github.com/btcsuite/goleveldb/leveldb" "github.com/btcsuite/goleveldb/leveldb/iterator" "github.com/btcsuite/goleveldb/leveldb/util" "github.com/p9c/p9/pkg/util/treap" ) const ( // defaultCacheSize is the default size for the database cache. defaultCacheSize = 64 * 1024 * 1024 // defaultFlushSecs is the default number of seconds to use as a threshold in between database cache flushes when // the cache size has not been exceeded. defaultFlushSecs = 36 // ldbBatchHeaderSize is the size of a leveldb batch header which includes the sequence header and record counter. ldbBatchHeaderSize = 12 // ldbRecordIKeySize is the size of the ikey used internally by leveldb when appending a record to a batch. ldbRecordIKeySize = 8 // These are used to help preallocate space needed for a batch in one allocation instead of letting leveldb itself // constantly grow it. // // This results in far less pressure on the GC and consequently helps prevent the GC from allocating a lot of extra // unneeded space. ) // ldbCacheIter wraps a treap iterator to provide the additional functionality needed to satisfy the // leveldb iterator.Iterator interface. type ldbCacheIter struct { *treap.Iterator } // Enforce ldbCacheIterator implements the leveldb iterator.Iterator interface. var _ iterator.Iterator = (*ldbCacheIter)(nil) // Error is only provided to satisfy the iterator interface as there are no errors for this memory-only structure. // // This is part of the leveldb iterator.Iterator interface implementation. func (iter *ldbCacheIter) Error() (e error) { return nil } // SetReleaser is only provided to satisfy the iterator interface as there is no need to override it. This is part of // the leveldb iterator. // // Iterator interface implementation. func (iter *ldbCacheIter) SetReleaser(releaser util.Releaser) { } // Release is only provided to satisfy the iterator interface. // // This is part of the leveldb iterator.Iterator interface implementation. func (iter *ldbCacheIter) Release() { } // newLdbCacheIter creates a new treap iterator for the given slice against the pending keys for the passed cache // snapshot and returns it wrapped in an ldbCacheIter so it can be used as a leveldb iterator. func newLdbCacheIter(snap *dbCacheSnapshot, slice *util.Range) *ldbCacheIter { iter := snap.pendingKeys.Iterator(slice.Start, slice.Limit) return &ldbCacheIter{Iterator: iter} } // dbCacheIterator defines an iterator over the key/value pairs in the database cache and underlying database. type dbCacheIterator struct { cacheSnapshot *dbCacheSnapshot dbIter iterator.Iterator cacheIter iterator.Iterator currentIter iterator.Iterator released bool } // Enforce dbCacheIterator implements the leveldb iterator.Iterator interface. var _ iterator.Iterator = (*dbCacheIterator)(nil) // skipPendingUpdates skips any keys at the current database iterator position that are being updated by the cache. // // The forwards flag indicates the direction the iterator is moving. func (iter *dbCacheIterator) skipPendingUpdates(forwards bool) { for iter.dbIter.Valid() { var skip bool key := iter.dbIter.Key() if iter.cacheSnapshot.pendingRemove.Has(key) { skip = true } else if iter.cacheSnapshot.pendingKeys.Has(key) { skip = true } if !skip { break } if forwards { iter.dbIter.Next() } else { iter.dbIter.Prev() } } } // chooseIterator first skips any entries in the database iterator that are being updated by the cache and sets the // current iterator to the appropriate iterator depending on their validity and the order they compare in while taking // into account the direction flag. // // When the iterator is being moved forwards and both iterators are valid, the iterator with the smaller key is chosen // and vice versa when the iterator is being moved backwards. func (iter *dbCacheIterator) chooseIterator(forwards bool) bool { // Skip any keys at the current database iterator position that are being updated by the cache. iter.skipPendingUpdates(forwards) // When both iterators are exhausted, the iterator is exhausted too. if !iter.dbIter.Valid() && !iter.cacheIter.Valid() { iter.currentIter = nil return false } // Choose the database iterator when the cache iterator is exhausted. if !iter.cacheIter.Valid() { iter.currentIter = iter.dbIter return true } // Choose the cache iterator when the database iterator is exhausted. if !iter.dbIter.Valid() { iter.currentIter = iter.cacheIter return true } // Both iterators are valid, so choose the iterator with either the smaller or larger key depending on the forwards // flag. compare := bytes.Compare(iter.dbIter.Key(), iter.cacheIter.Key()) if (forwards && compare > 0) || (!forwards && compare < 0) { iter.currentIter = iter.cacheIter } else { iter.currentIter = iter.dbIter } return true } // First positions the iterator at the first key/value pair and returns whether or not the pair exists. // // This is part of the leveldb iterator. Iterator interface implementation. func (iter *dbCacheIterator) First() bool { // Seek to the first key in both the database and cache iterators and choose the iterator that is both valid and has // the smaller key. iter.dbIter.First() iter.cacheIter.First() return iter.chooseIterator(true) } // Last positions the iterator at the last key/value pair and returns whether or not the pair exists. // // This is part of the leveldb iterator. Iterator interface implementation. func (iter *dbCacheIterator) Last() bool { // Seek to the last key in both the database and cache iterators and choose the iterator that is both valid and has // the larger key. iter.dbIter.Last() iter.cacheIter.Last() return iter.chooseIterator(false) } // Next moves the iterator one key/value pair forward and returns whether or not the pair exists. // // This is part of the leveldb iterator. Iterator interface implementation. func (iter *dbCacheIterator) Next() bool { // Nothing to return if cursor is exhausted. if iter.currentIter == nil { return false } // Move the current iterator to the next entry and choose the iterator that is both valid and has the smaller key. iter.currentIter.Next() return iter.chooseIterator(true) } // Prev moves the iterator one key/value pair backward and returns whether or not the pair exists. // // This is part of the leveldb iterator. Iterator interface implementation. func (iter *dbCacheIterator) Prev() bool { // Nothing to return if cursor is exhausted. if iter.currentIter == nil { return false } // Move the current iterator to the previous entry and choose the iterator that is both valid and has the larger // key. iter.currentIter.Prev() return iter.chooseIterator(false) } // Seek positions the iterator at the first key/value pair that is greater than or equal to the passed seek key. // // Returns false if no suitable key was found. // This is part of the leveldb iterator.Iterator interface implementation. func (iter *dbCacheIterator) Seek(key []byte) bool { // Seek to the provided key in both the database and cache iterators then choose the iterator that is both valid and // has the larger key. iter.dbIter.Seek(key) iter.cacheIter.Seek(key) return iter.chooseIterator(true) } // Valid indicates whether the iterator is positioned at a valid key/value pair. It will be considered invalid when the // iterator is newly created or exhausted. // // This is part of the leveldb iterator. Iterator interface implementation. func (iter *dbCacheIterator) Valid() bool { return iter.currentIter != nil } // Key returns the current key the iterator is pointing to. // // This is part of the leveldb iterator.Iterator interface implementation. func (iter *dbCacheIterator) Key() []byte { // Nothing to return if iterator is exhausted. if iter.currentIter == nil { return nil } return iter.currentIter.Key() } // Value returns the current value the iterator is pointing to. // // This is part of the leveldb iterator.Iterator interface implementation. func (iter *dbCacheIterator) Value() []byte { // Nothing to return if iterator is exhausted. if iter.currentIter == nil { return nil } return iter.currentIter.Value() } // SetReleaser is only provided to satisfy the iterator interface as there is // no need to override it. // // This is part of the leveldb iterator. Iterator interface implementation. func (iter *dbCacheIterator) SetReleaser(releaser util.Releaser) { } // Release releases the iterator by removing the underlying treap iterator from the list of active iterators against the // pending keys treap. // // This is part of the leveldb iterator.Iterator interface implementation. func (iter *dbCacheIterator) Release() { if !iter.released { iter.dbIter.Release() iter.cacheIter.Release() iter.currentIter = nil iter.released = true } } // Error is only provided to satisfy the iterator interface as there are no errors for this memory-only structure. // // This is part of the leveldb iterator.Iterator interface implementation. func (iter *dbCacheIterator) Error() (e error) { return nil } // dbCacheSnapshot defines a snapshot of the database cache and underlying database at a particular point in time. type dbCacheSnapshot struct { dbSnapshot *leveldb.Snapshot pendingKeys *treap.Immutable pendingRemove *treap.Immutable } // Has returns whether or not the passed key exists. func (snap *dbCacheSnapshot) Has(key []byte) bool { // Chk the cached entries first. if snap.pendingRemove.Has(key) { return false } if snap.pendingKeys.Has(key) { return true } // Consult the database. hasKey, _ := snap.dbSnapshot.Has(key, nil) return hasKey } // Get returns the value for the passed key. The function will return nil when the key does not exist. func (snap *dbCacheSnapshot) Get(key []byte) []byte { // Chk the cached entries first. if snap.pendingRemove.Has(key) { return nil } if value := snap.pendingKeys.Get(key); value != nil { return value } // Consult the database. value, e := snap.dbSnapshot.Get(key, nil) if e != nil { // F.Ln(err) return nil } return value } // Release releases the snapshot. func (snap *dbCacheSnapshot) Release() { snap.dbSnapshot.Release() snap.pendingKeys = nil snap.pendingRemove = nil } // NewIterator returns a new iterator for the snapshot. The newly returned iterator is not pointing to a valid item // until a call to one of the methods to position it is made. // // The slice parameter allows the iterator to be limited to a range of keys. // // The start key is inclusive and the limit key is exclusive. Either or both can be nil if the functionality is not // desired. func (snap *dbCacheSnapshot) NewIterator(slice *util.Range) *dbCacheIterator { return &dbCacheIterator{ dbIter: snap.dbSnapshot.NewIterator(slice, nil), cacheIter: newLdbCacheIter(snap, slice), cacheSnapshot: snap, } } // dbCache provides a database cache layer backed by an underlying database. // // It allows a maximum cache size and flush interval to be specified such that the cache is flushed to the database when // the cache size exceeds the maximum configured value or it has been longer than the configured interval since the last // flush. // // This effectively provides transaction batching so that callers can commit transactions at will without incurring // large performance hits due to frequent disk syncs. type dbCache struct { // ldb is the underlying leveldb DB for metadata. ldb *leveldb.DB // store is used to sync blocks to flat files. store *blockStore // The following fields are related to flushing the cache to persistent storage. // // Note that all flushing is performed in an opportunistic fashion. // // This means that it is only flushed during a transaction or when the database cache is closed. // // maxSize is the maximum size threshold the cache can grow to before it is flushed. // // flushInterval is the threshold interval of time that is allowed to pass before the cache is flushed. // // lastFlush is the time the cache was last flushed. It is used in conjunction with the current time and the flush // interval. // // NOTE: These flush related fields are protected by the database write lock. maxSize uint64 flushInterval time.Duration lastFlush time.Time // The following fields hold the keys that need to be stored or deleted from the underlying database once the cache // is full, enough time has passed, or when the database is shutting down. // // Note that these are stored using immutable treaps to support O(1) MVCC snapshots against the cached data. // // The cacheLock is used to protect concurrent access for cache updates and snapshots. cacheLock sync.RWMutex cachedKeys *treap.Immutable cachedRemove *treap.Immutable } // Snapshot returns a snapshot of the database cache and underlying database at a particular point in time. // // The snapshot must be released after use by calling Release. func (c *dbCache) Snapshot() (*dbCacheSnapshot, error) { dbSnapshot, e := c.ldb.GetSnapshot() if e != nil { str := "failed to open transaction" return nil, convertErr(str, e) } // Since the cached keys to be added and removed use an immutable treap, a snapshot is simply obtaining the root of // the tree under the lock which is used to atomically swap the root. c.cacheLock.RLock() cacheSnapshot := &dbCacheSnapshot{ dbSnapshot: dbSnapshot, pendingKeys: c.cachedKeys, pendingRemove: c.cachedRemove, } c.cacheLock.RUnlock() return cacheSnapshot, nil } // updateDB invokes the passed function in the context of a managed leveldb transaction. Any errors returned from the // user-supplied function will cause the transaction to be rolled back and are returned from this function. Otherwise, // the transaction is committed when the user-supplied function returns a nil error. func (c *dbCache) updateDB(fn func(ldbTx *leveldb.Transaction) error) (e error) { // Start a leveldb transaction. ldbTx, e := c.ldb.OpenTransaction() if e != nil { return convertErr("failed to open ldb transaction", e) } if e := fn(ldbTx); E.Chk(e) { ldbTx.Discard() return e } // Commit the leveldb transaction and convert any errors as needed. if e := ldbTx.Commit(); E.Chk(e) { return convertErr("failed to commit leveldb transaction", e) } return nil } // TreapForEacher is an interface which allows iteration of a treap in ascending order using a user-supplied callback // for each key/value pair. // // It mainly exists so both mutable and immutable treaps can be atomically committed to the database with the same // function. type TreapForEacher interface { ForEach(func(k, v []byte) bool) } // commitTreaps atomically commits all of the passed pending add/update/remove updates to the underlying database. func (c *dbCache) commitTreaps(pendingKeys, pendingRemove TreapForEacher) (e error) { // Perform all leveldb updates using an atomic transaction. return c.updateDB( func(ldbTx *leveldb.Transaction) (e error) { var innerErr error pendingKeys.ForEach( func(k, v []byte) bool { if dbErr := ldbTx.Put(k, v, nil); dbErr != nil { str := fmt.Sprintf( "failed to put key %q to "+ "ldb transaction", k, ) innerErr = convertErr(str, dbErr) return false } return true }, ) if innerErr != nil { return innerErr } pendingRemove.ForEach( func(k, v []byte) bool { if dbErr := ldbTx.Delete(k, nil); dbErr != nil { str := fmt.Sprintf( "failed to delete "+ "key %q from ldb transaction", k, ) innerErr = convertErr(str, dbErr) return false } return true }, ) return innerErr }, ) } // flush flushes the database cache to persistent storage. // // This involes syncing the block store and replaying all transactions that have been applied to the cache to the // underlying database. // // This function MUST be called with the database write lock held. func (c *dbCache) flush() (e error) { c.lastFlush = time.Now() // Sync the current write file associated with the block store. // // This is necessary before writing the metadata to prevent the case where the metadata contains information about a // block which actually hasn't been written yet in unexpected shutdown scenarios. if e := c.store.syncBlocks(); E.Chk(e) { return e } // Since the cached keys to be added and removed use an immutable treap, a snapshot is simply obtaining the root of // the tree under the lock which is used to atomically swap the root. c.cacheLock.RLock() cachedKeys := c.cachedKeys cachedRemove := c.cachedRemove c.cacheLock.RUnlock() // Nothing to do if there is no data to flush. if cachedKeys.Len() == 0 && cachedRemove.Len() == 0 { return nil } // Perform all leveldb updates using an atomic transaction. if e := c.commitTreaps(cachedKeys, cachedRemove); E.Chk(e) { return e } // Clear the cache since it has been flushed. c.cacheLock.Lock() c.cachedKeys = treap.NewImmutable() c.cachedRemove = treap.NewImmutable() c.cacheLock.Unlock() D.Ln("synced database to disk") return nil } // needsFlush returns whether or not the database cache needs to be flushed to persistent storage based on its current // size, whether or not adding all of the entries in the passed database transaction would cause it to exceed the // configured limit, and how much time has elapsed since the last time the cache was flushed. // // This function MUST be called with the database write lock held. func (c *dbCache) needsFlush(tx *transaction) bool { // A flush is needed when more time has elapsed than the configured flush interval. if time.Since(c.lastFlush) > c.flushInterval { return true } // A flush is needed when the size of the database cache exceeds the specified max cache size. // // The total calculated size is multiplied by 1. // // 5 here to account for additional memory consumption that will be needed during the flush as well as old nodes in // the cache that are referenced by the snapshot used by the transaction. snap := tx.snapshot totalSize := snap.pendingKeys.Size() + snap.pendingRemove.Size() totalSize = uint64(float64(totalSize) * 1.5) return totalSize > c.maxSize } // commitTx atomically adds all of the pending keys to add and remove into the database cache. // // When adding the pending keys would cause the size of the cache to exceed the max cache size, or the time since the // last flush exceeds the configured flush interval, the cache will be flushed to the underlying persistent database. // // This is an atomic operation with respect to the cache in that either all of the pending keys to add and remove in the // transaction will be applied or none of them will. The database cache itself might be flushed to the underlying // persistent database even if the transaction fails to apply, but it will only be the state of the cache without the // transaction applied. // // This function MUST be called during a database write transaction which in turn implies the database write lock will // be held. func (c *dbCache) commitTx(tx *transaction) (e error) { // Flush the cache and write the current transaction directly to the database if a flush is needed. if c.needsFlush(tx) { if e := c.flush(); E.Chk(e) { return e } // Perform all leveldb updates using an atomic transaction. e := c.commitTreaps(tx.pendingKeys, tx.pendingRemove) if e != nil { return e } // Clear the transaction entries since they have been committed. tx.pendingKeys = nil tx.pendingRemove = nil return nil } // At this point a database flush is not needed, so atomically commit the transaction to the cache. // // Since the cached keys to be added and removed use an immutable treap, a snapshot is simply obtaining the root of // the tree under the lock which is used to atomically swap the root. c.cacheLock.RLock() newCachedKeys := c.cachedKeys newCachedRemove := c.cachedRemove c.cacheLock.RUnlock() // Apply every key to add in the database transaction to the cache. tx.pendingKeys.ForEach( func(k, v []byte) bool { newCachedRemove = newCachedRemove.Delete(k) newCachedKeys = newCachedKeys.Put(k, v) return true }, ) tx.pendingKeys = nil // Apply every key to remove in the database transaction to the cache. tx.pendingRemove.ForEach( func(k, v []byte) bool { newCachedKeys = newCachedKeys.Delete(k) newCachedRemove = newCachedRemove.Put(k, nil) return true }, ) tx.pendingRemove = nil // Atomically replace the immutable treaps which hold the cached keys to add and delete. c.cacheLock.Lock() c.cachedKeys = newCachedKeys c.cachedRemove = newCachedRemove c.cacheLock.Unlock() return nil } // Close cleanly shuts down the database cache by syncing all data and closing the underlying leveldb database. // // This function MUST be called with the database write lock held. func (c *dbCache) Close() (e error) { // Flush any outstanding cached entries to disk. if e := c.flush(); E.Chk(e) { // Even if there is an error while flushing, attempt to close the underlying database. The error is ignored // since it would mask the flush error. _ = c.ldb.Close() return e } // Close the underlying leveldb database. if e := c.ldb.Close(); E.Chk(e) { str := "failed to close underlying leveldb database" return convertErr(str, e) } return nil } // newDbCache returns a new database cache instance backed by the provided leveldb instance. // // The cache will be flushed to leveldb when the max size exceeds the provided value or it has been longer than the // provided interval since the last flush. func newDbCache(ldb *leveldb.DB, store *blockStore, maxSize uint64, flushIntervalSecs uint32) *dbCache { return &dbCache{ ldb: ldb, store: store, maxSize: maxSize, flushInterval: time.Second * time.Duration(flushIntervalSecs), lastFlush: time.Now(), cachedKeys: treap.NewImmutable(), cachedRemove: treap.NewImmutable(), } }