Skip to content

refactor(sei-db): return cleanup closure from newTSReadOptions #2929

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pdrobnjak wants to merge 1 commit into
base: pd/fix-rocks-db-ci-fail
Choose a base branch
from
Open

refactor(sei-db): return cleanup closure from newTSReadOptions #2929

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 12 additions & 10 deletions sei-db/db_engine/rocksdb/mvcc/db.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,8 +127,8 @@ func OpenDB(dataDir string, config config.StateStoreConfig) (*Database, error) {
}

func (db *Database) getSlice(storeKey string, version int64, key []byte) (*grocksdb.Slice, error) {
readOpts := newTSReadOptions(version)
defer readOpts.Destroy()
readOpts, cleanup := newTSReadOptions(version)
defer cleanup()
return db.storage.GetCF(
readOpts,
db.cfHandle,
Expand Down Expand Up @@ -324,9 +324,9 @@ func (db *Database) Iterator(storeKey string, version int64, start, end []byte)
prefix := storePrefix(storeKey)
start, end = util.IterateWithPrefix(prefix, start, end)

readOpts := newTSReadOptions(version)
readOpts, cleanup := newTSReadOptions(version)
itr := db.storage.NewIteratorCF(readOpts, db.cfHandle)
return NewRocksDBIterator(itr, readOpts, prefix, start, end, version, db.earliestVersion, false), nil
return NewRocksDBIterator(itr, cleanup, prefix, start, end, version, db.earliestVersion, false), nil
}

func (db *Database) ReverseIterator(storeKey string, version int64, start, end []byte) (types.DBIterator, error) {
Expand All @@ -341,9 +341,9 @@ func (db *Database) ReverseIterator(storeKey string, version int64, start, end [
prefix := storePrefix(storeKey)
start, end = util.IterateWithPrefix(prefix, start, end)

readOpts := newTSReadOptions(version)
readOpts, cleanup := newTSReadOptions(version)
itr := db.storage.NewIteratorCF(readOpts, db.cfHandle)
return NewRocksDBIterator(itr, readOpts, prefix, start, end, version, db.earliestVersion, true), nil
return NewRocksDBIterator(itr, cleanup, prefix, start, end, version, db.earliestVersion, true), nil
}

// Import loads the initial version of the state in parallel with numWorkers goroutines
Expand Down Expand Up @@ -416,7 +416,7 @@ func (db *Database) RawIterate(storeKey string, fn func(key []byte, value []byte
readOpts.SetTimestamp(endTs[:])

itr := db.storage.NewIteratorCF(readOpts, db.cfHandle)
rocksItr := NewRocksDBIterator(itr, readOpts, prefix, start, end, latestVersion, 1, false)
rocksItr := NewRocksDBIterator(itr, func() { readOpts.Destroy() }, prefix, start, end, latestVersion, 1, false)
defer func() { _ = rocksItr.Close() }()

for rocksItr.Valid() {
Expand All @@ -443,15 +443,17 @@ func (db *Database) GetLatestMigratedModule() (string, error) {
panic("not implemented")
}

// newTSReadOptions returns ReadOptions used in the RocksDB column family read.
func newTSReadOptions(version int64) *grocksdb.ReadOptions {
// newTSReadOptions returns ReadOptions used in the RocksDB column family read
// and a cleanup function that destroys them. The caller must ensure cleanup is
// called when the ReadOptions are no longer needed.
func newTSReadOptions(version int64) (*grocksdb.ReadOptions, func()) {
var ts [TimestampSize]byte
binary.LittleEndian.PutUint64(ts[:], uint64(version))

readOpts := grocksdb.NewDefaultReadOptions()
readOpts.SetTimestamp(ts[:])

return readOpts
return readOpts, func() { readOpts.Destroy() }
}

func storePrefix(storeKey string) []byte {
Expand Down
42 changes: 21 additions & 21 deletions sei-db/db_engine/rocksdb/mvcc/iterator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,25 +14,25 @@ var _ types.DBIterator = (*iterator)(nil)

type iterator struct {
source *grocksdb.Iterator
readOpts *grocksdb.ReadOptions
cleanup func()
prefix, start, end []byte
version int64
reverse bool
invalid bool
}

func NewRocksDBIterator(source *grocksdb.Iterator, readOpts *grocksdb.ReadOptions, prefix, start, end []byte, version int64, earliestVersion int64, reverse bool) *iterator {
func NewRocksDBIterator(source *grocksdb.Iterator, cleanup func(), prefix, start, end []byte, version int64, earliestVersion int64, reverse bool) *iterator {
// Return invalid iterator if requested iterator height is lower than earliest version after pruning
if version < earliestVersion {
return &iterator{
source: source,
readOpts: readOpts,
prefix: prefix,
start: start,
end: end,
version: version,
reverse: reverse,
invalid: true,
source: source,
cleanup: cleanup,
prefix: prefix,
start: start,
end: end,
version: version,
reverse: reverse,
invalid: true,
}
}

Expand Down Expand Up @@ -60,14 +60,14 @@ func NewRocksDBIterator(source *grocksdb.Iterator, readOpts *grocksdb.ReadOption
}

return &iterator{
source: source,
readOpts: readOpts,
prefix: prefix,
start: start,
end: end,
version: version,
reverse: reverse,
invalid: !source.Valid(),
source: source,
cleanup: cleanup,
prefix: prefix,
start: start,
end: end,
version: version,
reverse: reverse,
invalid: !source.Valid(),
}
}

Expand Down Expand Up @@ -160,9 +160,9 @@ func (itr *iterator) Error() error {
func (itr *iterator) Close() error {
itr.source.Close()
itr.source = nil
if itr.readOpts != nil {
itr.readOpts.Destroy()
itr.readOpts = nil
if itr.cleanup != nil {
itr.cleanup()
itr.cleanup = nil
}
itr.invalid = true
return nil
Expand Down
Loading