You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "sashapolo (via GitHub)" <gi...@apache.org> on 2023/06/02 06:38:17 UTC
[GitHub] [ignite-3] sashapolo commented on a diff in pull request #2126: IGNITE-19367 Remove MvPartitionStorage#pollForVacuum
sashapolo commented on code in PR #2126:
URL: https://github.com/apache/ignite-3/pull/2126#discussion_r1213969894
##########
modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BaseMvPartitionStorageTest.java:
##########
@@ -192,8 +193,29 @@ protected BinaryRow abortWrite(RowId rowId) {
});
}
- protected BinaryRowAndRowId pollForVacuum(HybridTimestamp lowWatermark) {
- //TODO IGNITE-19367 Remove or replace with some other method.
- return storage.pollForVacuum(lowWatermark);
+ @Nullable BinaryRowAndRowId pollForVacuum(HybridTimestamp lowWatermark) {
Review Comment:
Please add a javadoc
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java:
##########
@@ -65,56 +65,59 @@ public PendingComparableValuesTracker<HybridTimestamp, Void> getSafeTimeTracker(
/**
* Tries removing {@code count} oldest stale entries and their indexes.
- * If there's less entries that can be removed, then exits prematurely.
+ * If there are fewer rows than the {@code count}, or it was not possible to lock any, then exits prematurely.
*
* @param lowWatermark Low watermark for the vacuum.
* @param count Count of entries to GC.
+ * @return {@code False} if there is no garbage left in the storage.
*/
- public void vacuumBatch(HybridTimestamp lowWatermark, int count) {
- for (int i = 0; i < count; i++) {
- if (!storage.runConsistently(locker -> internalVacuum(lowWatermark))) {
- break;
+ public boolean vacuumBatch(HybridTimestamp lowWatermark, int count) {
+ return storage.runConsistently(locker -> {
+ for (int i = 0; i < count; i++) {
+ if (!internalVacuum(lowWatermark, locker)) {
+ return false;
+ }
}
- }
- }
- /**
- * Tries removing partition's oldest stale entry and its indexes.
- *
- * @param lowWatermark Low watermark for the vacuum.
- * @return {@code true} if an entry was garbage collected, {@code false} if there was nothing to collect.
- * @see MvPartitionStorage#pollForVacuum(HybridTimestamp)
- */
- public boolean vacuum(HybridTimestamp lowWatermark) {
- return storage.runConsistently(locker -> internalVacuum(lowWatermark));
+ return true;
+ });
}
/**
- * Executes garbage collection.
+ * Attempts to collect garbage for one {@link RowId}, if it fails to lock it, then immediately breaks.
Review Comment:
```suggestion
* Attempts to collect garbage for one {@link RowId}, if it fails to lock it, then immediately stops.
```
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/AbstractMvStorageUpdateHandlerTest.java:
##########
@@ -131,7 +131,7 @@ void tearDown() throws Exception {
@Test
void testConcurrentExecuteBatchGc() {
- assertThat(distributionZoneConfig.dataStorage().gcOnUpdateBatchSize().update(2), willSucceedFast());
+ assertThat(distributionZoneConfig.dataStorage().gcOnUpdateBatchSize().update(4), willSucceedFast());
Review Comment:
What's this change for?
##########
modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/GarbageCollector.java:
##########
@@ -205,7 +205,10 @@ boolean tryAddToGcQueue(WriteBatchWithIndex writeBatch, RowId rowId, HybridTimes
// However, the element that we need to garbage collect is the next (older one) element.
// First we check if there's anything to garbage collect. If the element is a tombstone we remove it.
// If the next element exists, that should be the element that we want to garbage collect.
- try (RocksIterator gcIt = db.newIterator(gcQueueCf, helper.upperBoundReadOpts)) {
+ try (
+ RocksIterator newGcIt = db.newIterator(gcQueueCf, helper.upperBoundReadOpts);
Review Comment:
This pattern can be seen a lot of times in this code, I think it should be extracted into a separate method
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java:
##########
@@ -65,56 +65,59 @@ public PendingComparableValuesTracker<HybridTimestamp, Void> getSafeTimeTracker(
/**
* Tries removing {@code count} oldest stale entries and their indexes.
- * If there's less entries that can be removed, then exits prematurely.
+ * If there are fewer rows than the {@code count}, or it was not possible to lock any, then exits prematurely.
*
* @param lowWatermark Low watermark for the vacuum.
* @param count Count of entries to GC.
+ * @return {@code False} if there is no garbage left in the storage.
*/
- public void vacuumBatch(HybridTimestamp lowWatermark, int count) {
- for (int i = 0; i < count; i++) {
- if (!storage.runConsistently(locker -> internalVacuum(lowWatermark))) {
- break;
+ public boolean vacuumBatch(HybridTimestamp lowWatermark, int count) {
+ return storage.runConsistently(locker -> {
+ for (int i = 0; i < count; i++) {
+ if (!internalVacuum(lowWatermark, locker)) {
+ return false;
+ }
}
- }
- }
- /**
- * Tries removing partition's oldest stale entry and its indexes.
- *
- * @param lowWatermark Low watermark for the vacuum.
- * @return {@code true} if an entry was garbage collected, {@code false} if there was nothing to collect.
- * @see MvPartitionStorage#pollForVacuum(HybridTimestamp)
- */
- public boolean vacuum(HybridTimestamp lowWatermark) {
- return storage.runConsistently(locker -> internalVacuum(lowWatermark));
+ return true;
+ });
}
/**
- * Executes garbage collection.
+ * Attempts to collect garbage for one {@link RowId}, if it fails to lock it, then immediately breaks.
*
- * <p>Must be called inside a {@link MvPartitionStorage#runConsistently(WriteClosure)} closure.
+ * <p>Must be called inside a {@link PartitionDataStorage#runConsistently(WriteClosure)} closure.
*
* @param lowWatermark Low watermark for the vacuum.
- * @return {@code true} if an entry was garbage collected, {@code false} if there was nothing to collect.
+ * @param locker From {@link PartitionDataStorage#runConsistently(WriteClosure)}.
+ * @return {@code False} if there is no garbage left in the {@link #storage}.
*/
- private boolean internalVacuum(HybridTimestamp lowWatermark) {
- BinaryRowAndRowId vacuumed = storage.pollForVacuum(lowWatermark);
+ private boolean internalVacuum(HybridTimestamp lowWatermark, Locker locker) {
+ while (true) {
+ GcEntry gcEntry = storage.peek(lowWatermark);
- if (vacuumed == null) {
- // Nothing was garbage collected.
- return false;
- }
+ if (gcEntry == null) {
+ return false;
+ }
- BinaryRow binaryRow = vacuumed.binaryRow();
+ RowId rowId = gcEntry.getRowId();
- assert binaryRow != null;
+ if (!locker.tryLock(rowId)) {
Review Comment:
Why do we stop if we are unable to lock a row? What should the user do with this information? I think this should be reflected in the javadoc
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandlerTest.java:
##########
@@ -55,20 +55,24 @@ void testVacuum() {
HybridTimestamp lowWatermark = new HybridTimestamp(100, 100);
- assertFalse(gcUpdateHandler.vacuum(lowWatermark));
- verify(partitionStorage).pollForVacuum(lowWatermark);
+ assertFalse(gcUpdateHandler.vacuumBatch(lowWatermark, 1));
Review Comment:
Looks like we don't have any tests that use a batch size larger than 1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org