Skip to content
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

Assert DB is open before createStreamKeyRaw #9014

Closed
wants to merge 17 commits into from
Closed

Assert DB is open before createStreamKeyRaw #9014

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
342718e
asset DB is open before createStreamKeyRaw
gfukushima Jan 20, 2025
78c0875
asset DB is open before createStreamRaw
gfukushima Jan 20, 2025
9b7c189
asset DB is open before createStreamKeyRaw
gfukushima Jan 20, 2025
2d8713b
stop blob and state pruner as well
gfukushima Jan 21, 2025
c516e93
try and close columnHandle before closing db
gfukushima Jan 21, 2025
3d3182c
log resource name
gfukushima Jan 21, 2025
a90192b
spotless
gfukushima Jan 21, 2025
4a1abe9
check if RocksDbStats are open before providing the external summary
gfukushima Jan 22, 2025
4185b9d
spotless
gfukushima Jan 22, 2025
02e3c72
Merge branch 'master' into rocksdb-sigsevv-fix
gfukushima Jan 23, 2025
376577f
close iterators after streaming
gfukushima Jan 23, 2025
c5d6892
spotless
gfukushima Jan 23, 2025
b12d493
Merge remote-tracking branch 'origin/rocksdb-sigsevv-fix' into rocksd…
gfukushima Jan 23, 2025
22ea2a7
shutdown asyncRunners on shutdown
gfukushima Jan 23, 2025
c514158
cancel background work before closing db
gfukushima Jan 23, 2025
ab108ac
add debug logs
gfukushima Jan 23, 2025
23f1d8f
cancel all bg work
gfukushima Jan 24, 2025
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
9 changes: 6 additions & 3 deletions ...es/chainstorage/src/main/java/tech/pegasys/teku/services/chainstorage/StorageService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,9 +271,12 @@ public Optional<StatePruner> getStatePruner() {

@Override
protected SafeFuture<?> doStop() {
return blockPruner
.map(BlockPruner::stop)
.orElseGet(() -> SafeFuture.completedFuture(null))
return SafeFuture.allOf(
blockPruner.map(BlockPruner::stop).orElseGet(() -> SafeFuture.completedFuture(null)),
blobsPruner
.map(BlobSidecarPruner::stop)
.orElseGet(() -> SafeFuture.completedFuture(null)),
statePruner.map(StatePruner::stop).orElseGet(() -> SafeFuture.completedFuture(null)))
gfukushima marked this conversation as resolved.
Show resolved Hide resolved
.thenCompose(__ -> SafeFuture.fromRunnable(database::close));
}

Expand Down
3 changes: 3 additions & 0 deletions storage/src/main/java/tech/pegasys/teku/storage/server/pruner/BlobSidecarPruner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,10 @@ protected synchronized SafeFuture<?> doStart() {

@Override
protected synchronized SafeFuture<?> doStop() {
LOG.debug("Stopping blob pruner");
asyncRunner.shutdown();
scheduledPruner.ifPresent(Cancellable::cancel);

return SafeFuture.COMPLETE;
}

Expand Down
2 changes: 2 additions & 0 deletions storage/src/main/java/tech/pegasys/teku/storage/server/pruner/BlockPruner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,8 @@ protected synchronized SafeFuture<?> doStart() {

@Override
protected synchronized SafeFuture<?> doStop() {
LOG.debug("Stopping block pruner");
asyncRunner.shutdown();
scheduledPruner.ifPresent(Cancellable::cancel);
return SafeFuture.COMPLETE;
}
Expand Down
2 changes: 2 additions & 0 deletions storage/src/main/java/tech/pegasys/teku/storage/server/pruner/StatePruner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,8 @@ protected synchronized SafeFuture<?> doStart() {

@Override
protected synchronized SafeFuture<?> doStop() {
LOG.debug("Stopping state pruner");
asyncRunner.shutdown();
scheduledPruner.ifPresent(Cancellable::cancel);
return SafeFuture.COMPLETE;
}
Expand Down
18 changes: 16 additions & 2 deletions storage/src/main/java/tech/pegasys/teku/storage/server/rocksdb/RocksDbInstance.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.tuweni.bytes.Bytes;
import org.rocksdb.AbstractRocksIterator;
import org.rocksdb.ColumnFamilyHandle;
Expand All @@ -47,6 +49,8 @@ public class RocksDbInstance implements KvStoreAccessor {

private final AtomicBoolean closed = new AtomicBoolean(false);

private static final Logger LOG = LogManager.getLogger();

RocksDbInstance(
final TransactionDB db,
final ColumnFamilyHandle defaultHandle,
Expand All @@ -60,6 +64,7 @@ public class RocksDbInstance implements KvStoreAccessor {

@Override
public <T> Optional<T> get(final KvStoreVariable<T> variable) {
assertOpen();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this one shouldnt be needed? it's asserted in getRaw...

return getRaw(variable).map(data -> variable.getSerializer().deserialize(data.toArrayUnsafe()));
}

Expand Down Expand Up @@ -160,6 +165,7 @@ public Stream<ColumnEntry<Bytes, Bytes>> streamRaw(final KvStoreColumn<?, ?> col
@Override
@MustBeClosed
public Stream<Bytes> streamKeysRaw(final KvStoreColumn<?, ?> column) {
assertOpen();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is checked in createStreamRaw, we can let it go through here i think...

return createStreamKeyRaw(column, RocksIterator::seekToFirst, key -> true).map(Bytes::wrap);
}

Expand Down Expand Up @@ -249,10 +255,13 @@ private <K, V> Stream<ColumnEntry<byte[], byte[]>> createStreamRaw(
final KvStoreColumn<K, V> column,
final Consumer<RocksIterator> setupIterator,
final Predicate<K> continueTest) {
assertOpen();
final ColumnFamilyHandle handle = columnHandles.get(column);
final RocksIterator rocksDbIterator = db.newIterator(handle);
setupIterator.accept(rocksDbIterator);
return RocksDbIterator.create(column, rocksDbIterator, continueTest, closed::get).toStream();
return RocksDbIterator.create(column, rocksDbIterator, continueTest, closed::get)
.toStream()
.onClose(rocksDbIterator::close);
}

@SuppressWarnings("MustBeClosedChecker")
Expand All @@ -261,10 +270,13 @@ private <K, V> Stream<byte[]> createStreamKeyRaw(
final KvStoreColumn<K, V> column,
final Consumer<RocksIterator> setupIterator,
final Predicate<K> continueTest) {
assertOpen();
final ColumnFamilyHandle handle = columnHandles.get(column);
final RocksIterator rocksDbIterator = db.newIterator(handle);
setupIterator.accept(rocksDbIterator);
return RocksDbKeyIterator.create(column, rocksDbIterator, continueTest, closed::get).toStream();
return RocksDbKeyIterator.create(column, rocksDbIterator, continueTest, closed::get)
.toStream()
.onClose(rocksDbIterator::close);
}

@Override
Expand All @@ -274,7 +286,9 @@ public synchronized void close() throws Exception {
openTransaction.closeViaDatabase();
}
db.syncWal();
db.cancelAllBackgroundWork(false);
for (final AutoCloseable resource : resources) {
LOG.debug("Closing resource: {}", resource);
resource.close();
}
}
Expand Down
2 changes: 1 addition & 1 deletion storage/src/main/java/tech/pegasys/teku/storage/server/rocksdb/RocksDbInstanceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,7 +189,7 @@ private static BlockBasedTableConfig createBlockBasedTableConfig(final Cache cac
.setBlockCache(cache)
.setFilterPolicy(new BloomFilter(10, false))
.setPartitionFilters(true)
.setCacheIndexAndFilterBlocks(true)
.setCacheIndexAndFilterBlocks(false)
.setBlockSize(ROCKSDB_BLOCK_SIZE);
}
}
6 changes: 5 additions & 1 deletion storage/src/main/java/tech/pegasys/teku/storage/server/rocksdb/RocksDbStats.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

package tech.pegasys.teku.storage.server.rocksdb;

import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.function.Supplier;
Expand Down Expand Up @@ -195,7 +196,10 @@ public void registerMetrics(final RocksDB database) {
+ "_"
+ histogram.name().toLowerCase(Locale.ROOT),
"RocksDB histogram for " + histogram.name(),
() -> provideExternalSummary(stats, histogram));
() ->
ifOpen(
() -> provideExternalSummary(stats, histogram),
new ExternalSummary(0, 0, Collections.emptyList())));
}
}
}
Expand Down
Loading