You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by mk...@apache.org on 2022/09/16 13:12:33 UTC
[geode] branch support/1.15 updated (62e60b5062 -> d1910cf1ee)
This is an automated email from the ASF dual-hosted git repository.
mkevo pushed a change to branch support/1.15
in repository https://gitbox.apache.org/repos/asf/geode.git
from 62e60b5062 GEODE-10420: Finish distribute() work if interrupted (#7854)
new bc47b30c0b GEODE-10412: Clear expired tombstones during region destroy (#7838)
new d1910cf1ee GEODE-10401: Configurable .drf recovery HashMap overflow threshold (#7828)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../cache/versions/TombstoneDUnitTest.java | 42 +++++++++++++++
.../apache/geode/internal/cache/DiskStoreImpl.java | 61 +++++++++++++++------
.../geode/internal/cache/TombstoneService.java | 6 ++-
.../OplogEntryIdSetDrfHashSetThresholdTest.java | 62 ++++++++++++++++++++++
.../geode/internal/cache/TombstoneServiceTest.java | 46 +++++++++++++---
5 files changed, 193 insertions(+), 24 deletions(-)
create mode 100644 geode-core/src/test/java/org/apache/geode/internal/cache/OplogEntryIdSetDrfHashSetThresholdTest.java
[geode] 01/02: GEODE-10412: Clear expired tombstones during region destroy (#7838)
Posted by mk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mkevo pushed a commit to branch support/1.15
in repository https://gitbox.apache.org/repos/asf/geode.git
commit bc47b30c0bb57a9c9437234a8fc8530f1daf14bc
Author: Jakov Varenina <62...@users.noreply.github.com>
AuthorDate: Mon Aug 29 17:08:26 2022 +0200
GEODE-10412: Clear expired tombstones during region destroy (#7838)
* GEODE-10412: Clear expired tombstones during region destroy
The issue:
During region destroy operation, the expired tombstones aren't cleared
when non-expired ones are available. Later, these expired
tombstones prevent all other regions' tombstones from being cleared
from memory, causing many issues (memory and disk exhaustion).
The solution:
When a region is destroyed, it must clear all the related expired and
non-expired tombstones from memory.
* Add distributed test that reproduce the issue
* Update after review
---
.../cache/versions/TombstoneDUnitTest.java | 42 ++++++++++++++++++++
.../geode/internal/cache/TombstoneService.java | 6 ++-
.../geode/internal/cache/TombstoneServiceTest.java | 46 ++++++++++++++++++----
3 files changed, 86 insertions(+), 8 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/versions/TombstoneDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/versions/TombstoneDUnitTest.java
index bbcf0ca6ec..d4bd0cc89b 100644
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/versions/TombstoneDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/versions/TombstoneDUnitTest.java
@@ -14,11 +14,13 @@
*/
package org.apache.geode.internal.cache.versions;
+import static org.apache.geode.cache.RegionShortcut.PARTITION_PERSISTENT;
import static org.apache.geode.cache.RegionShortcut.REPLICATE;
import static org.apache.geode.cache.RegionShortcut.REPLICATE_PERSISTENT;
import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
import static org.apache.geode.internal.cache.InitialImageOperation.GIITestHookType.DuringApplyDelta;
import static org.apache.geode.internal.cache.InitialImageOperation.resetAllGIITestHooks;
+import static org.apache.geode.internal.cache.TombstoneService.EXPIRED_TOMBSTONE_LIMIT;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
@@ -121,6 +123,35 @@ public class TombstoneDUnitTest implements Serializable {
});
}
+ @Test
+ public void testTombstoneExpiredAndNonExpiredAreClearedAfterRegionIsDestroyed() {
+ VM vm0 = VM.getVM(0);
+
+ vm0.invoke(() -> {
+ // reduce timeout so that tombstone is immediately marked as expired
+ TombstoneService.REPLICATE_TOMBSTONE_TIMEOUT = 100;
+ createCacheAndRegion(PARTITION_PERSISTENT);
+ region.put("K1", "V1");
+ region.destroy("K1");
+ });
+
+ vm0.invoke(() -> {
+ waitForScheduledTombstoneCount(0);
+ // increase timeout so that next tombstone doesn't expire
+ TombstoneService.REPLICATE_TOMBSTONE_TIMEOUT = 150000;
+ region.put("K1", "V1");
+ region.destroy("K1");
+
+ region.destroyRegion();
+ // force expiry of batch - there is only one expired tombstone at this moment
+ EXPIRED_TOMBSTONE_LIMIT = 1;
+ });
+
+ vm0.invoke(() -> {
+ createCacheAndRegion(PARTITION_PERSISTENT);
+ checkExpiredTombstones(0);
+ });
+ }
@Test
public void testWhenAnOutOfRangeTimeStampIsSeenWeExpireItInReplicateTombstoneSweeper() {
@@ -562,6 +593,17 @@ public class TombstoneDUnitTest implements Serializable {
}
}
+ private void waitForScheduledTombstoneCount(int count) {
+ LocalRegion region = (LocalRegion) cache.getRegion(REGION_NAME);
+ await().until(() -> ((InternalCache) cache).getTombstoneService().getSweeper(region).tombstones
+ .size() == count);
+ }
+
+ private void checkExpiredTombstones(int count) {
+ await().until(
+ () -> ((InternalCache) cache).getTombstoneService().getScheduledTombstoneCount() == count);
+ }
+
private void performGC(int count) throws Exception {
((InternalCache) cache).getTombstoneService().forceBatchExpirationForTests(count);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TombstoneService.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TombstoneService.java
index 242a3ff20b..dc3532a734 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TombstoneService.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TombstoneService.java
@@ -926,7 +926,11 @@ public class TombstoneService {
* @return true if predicate ever returned true
*/
private boolean removeIf(Predicate<Tombstone> predicate) {
- return removeUnexpiredIf(predicate) || removeExpiredIf(predicate);
+ boolean isTombstoneRemoved = removeUnexpiredIf(predicate);
+ if (removeExpiredIf(predicate)) {
+ isTombstoneRemoved = true;
+ }
+ return isTombstoneRemoved;
}
synchronized void start() {
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/TombstoneServiceTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/TombstoneServiceTest.java
index 37bdfd63c3..9e6c437a82 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/TombstoneServiceTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/TombstoneServiceTest.java
@@ -14,6 +14,7 @@
*/
package org.apache.geode.internal.cache;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -40,7 +41,9 @@ public class TombstoneServiceTest {
DistributedRegion region;
VersionTag destroyedVersion;
private TombstoneService.ReplicateTombstoneSweeper replicateTombstoneSweeper;
- private TombstoneService.Tombstone tombstone;
+ private TombstoneService.Tombstone tombstone1;
+
+ private TombstoneService.Tombstone tombstone2;
@Before
@@ -55,8 +58,9 @@ public class TombstoneServiceTest {
destroyedVersion = mock(VersionTag.class);
replicateTombstoneSweeper = new TombstoneService.ReplicateTombstoneSweeper(cacheTime, stats,
cancelCriterion, executor);
- tombstone = new TombstoneService.Tombstone(entry, region, destroyedVersion);
- tombstone.entry = entry;
+ tombstone1 = new TombstoneService.Tombstone(entry, region, destroyedVersion);
+ tombstone2 = new TombstoneService.Tombstone(entry, region, destroyedVersion);
+ tombstone1.entry = entry;
}
@Test
@@ -64,9 +68,9 @@ public class TombstoneServiceTest {
when(region.isInitialized()).thenReturn(false);
when(region.getRegionMap()).thenReturn(regionMap);
- replicateTombstoneSweeper.expireTombstone(tombstone);
+ replicateTombstoneSweeper.expireTombstone(tombstone1);
replicateTombstoneSweeper.expireBatch();
- verify(regionMap, Mockito.never()).removeTombstone(tombstone.entry, tombstone);
+ verify(regionMap, Mockito.never()).removeTombstone(tombstone1.entry, tombstone1);
}
@Test
@@ -80,8 +84,36 @@ public class TombstoneServiceTest {
when(region.getDiskRegion()).thenReturn(mock(DiskRegion.class));
- replicateTombstoneSweeper.expireTombstone(tombstone);
+ replicateTombstoneSweeper.expireTombstone(tombstone1);
replicateTombstoneSweeper.expireBatch();
- verify(regionMap).removeTombstone(tombstone.entry, tombstone);
+ verify(regionMap).removeTombstone(tombstone1.entry, tombstone1);
+ }
+
+ @Test
+ public void validateThatTheExpiredTombstonesAreCleared() {
+ when(region.getRegionMap()).thenReturn(regionMap);
+ replicateTombstoneSweeper.expireTombstone(tombstone1);
+ assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isOne();
+ replicateTombstoneSweeper.unscheduleTombstones(region);
+ assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isZero();
+ }
+
+ @Test
+ public void validateThatTheNonExpiredTombstonesAreCleared() {
+ when(region.getRegionMap()).thenReturn(regionMap);
+ replicateTombstoneSweeper.scheduleTombstone(tombstone1);
+ assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isOne();
+ replicateTombstoneSweeper.unscheduleTombstones(region);
+ assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isZero();
+ }
+
+ @Test
+ public void validateThatTheNonExpiredAndExpiredTombstonesAreCleared() {
+ when(region.getRegionMap()).thenReturn(regionMap);
+ replicateTombstoneSweeper.scheduleTombstone(tombstone1);
+ replicateTombstoneSweeper.expireTombstone(tombstone2);
+ assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isEqualTo(2);
+ replicateTombstoneSweeper.unscheduleTombstones(region);
+ assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isZero();
}
}
[geode] 02/02: GEODE-10401: Configurable .drf recovery HashMap overflow threshold (#7828)
Posted by mk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mkevo pushed a commit to branch support/1.15
in repository https://gitbox.apache.org/repos/asf/geode.git
commit d1910cf1ee18aae749da3553b3537d74b7ef5cae
Author: Jakov Varenina <62...@users.noreply.github.com>
AuthorDate: Wed Jul 27 15:08:53 2022 +0200
GEODE-10401: Configurable .drf recovery HashMap overflow threshold (#7828)
Configurable with the jvm parameter:
gemfire.disk.drfHashMapOverflowThreshold
Default value: 805306368
When configured threshold value is reached, then server will overflow to
the new hashmap during the recovery of .drf files. Warning: If you set
threshold parameter over 805306368, then uneeded delay will happen due
to bug in fastutil dependency.
---
.../apache/geode/internal/cache/DiskStoreImpl.java | 61 +++++++++++++++------
.../OplogEntryIdSetDrfHashSetThresholdTest.java | 62 ++++++++++++++++++++++
2 files changed, 107 insertions(+), 16 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DiskStoreImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DiskStoreImpl.java
index 9dee1c1c77..b544ca065f 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DiskStoreImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DiskStoreImpl.java
@@ -180,6 +180,13 @@ public class DiskStoreImpl implements DiskStore {
public static final String RECOVER_VALUES_SYNC_PROPERTY_NAME =
GeodeGlossary.GEMFIRE_PREFIX + "disk.recoverValuesSync";
+ /**
+ * When configured threshold value is reached, then server will overflow to
+ * the new hashmap during the recovery of .drf files
+ */
+ public static final String DRF_HASHMAP_OVERFLOW_THRESHOLD_NAME =
+ GeodeGlossary.GEMFIRE_PREFIX + "disk.drfHashMapOverflowThreshold";
+
/**
* Allows recovering values for LRU regions. By default values are not recovered for LRU regions
* during recovery.
@@ -187,6 +194,10 @@ public class DiskStoreImpl implements DiskStore {
public static final String RECOVER_LRU_VALUES_PROPERTY_NAME =
GeodeGlossary.GEMFIRE_PREFIX + "disk.recoverLruValues";
+ static final long DRF_HASHMAP_OVERFLOW_THRESHOLD_DEFAULT = 805306368;
+ static final long DRF_HASHMAP_OVERFLOW_THRESHOLD =
+ Long.getLong(DRF_HASHMAP_OVERFLOW_THRESHOLD_NAME, DRF_HASHMAP_OVERFLOW_THRESHOLD_DEFAULT);
+
boolean RECOVER_VALUES = getBoolean(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, true);
boolean RECOVER_VALUES_SYNC = getBoolean(DiskStoreImpl.RECOVER_VALUES_SYNC_PROPERTY_NAME, false);
@@ -3525,31 +3536,49 @@ public class DiskStoreImpl implements DiskStore {
}
try {
- if (id > 0 && id <= 0x00000000FFFFFFFFL) {
- currentInts.get().add((int) id);
+ if (shouldOverflow(id)) {
+ overflowToNewHashMap(id);
} else {
- currentLongs.get().add(id);
+ if (id > 0 && id <= 0x00000000FFFFFFFFL) {
+ this.currentInts.get().add((int) id);
+ } else {
+ this.currentLongs.get().add(id);
+ }
}
} catch (IllegalArgumentException illegalArgumentException) {
// See GEODE-8029.
- // Too many entries on the accumulated drf files, overflow and continue.
+ // Too many entries on the accumulated drf files, overflow next [Int|Long]OpenHashSet and
+ // continue.
+ overflowToNewHashMap(id);
+ }
+ }
+
+ boolean shouldOverflow(final long id) {
+ if (id > 0 && id <= 0x00000000FFFFFFFFL) {
+ return currentInts.get().size() == DRF_HASHMAP_OVERFLOW_THRESHOLD;
+ } else {
+ return currentLongs.get().size() == DRF_HASHMAP_OVERFLOW_THRESHOLD;
+ }
+ }
+
+ void overflowToNewHashMap(final long id) {
+ if (DRF_HASHMAP_OVERFLOW_THRESHOLD == DRF_HASHMAP_OVERFLOW_THRESHOLD_DEFAULT) {
logger.warn(
"There is a large number of deleted entries within the disk-store, please execute an offline compaction.");
+ }
- // Overflow to the next [Int|Long]OpenHashSet and continue.
- if (id > 0 && id <= 0x00000000FFFFFFFFL) {
- IntOpenHashSet overflownHashSet = new IntOpenHashSet((int) INVALID_ID);
- allInts.add(overflownHashSet);
- currentInts.set(overflownHashSet);
+ if (id > 0 && id <= 0x00000000FFFFFFFFL) {
+ IntOpenHashSet overflownHashSet = new IntOpenHashSet((int) INVALID_ID);
+ allInts.add(overflownHashSet);
+ currentInts.set(overflownHashSet);
- currentInts.get().add((int) id);
- } else {
- LongOpenHashSet overflownHashSet = new LongOpenHashSet((int) INVALID_ID);
- allLongs.add(overflownHashSet);
- currentLongs.set(overflownHashSet);
+ currentInts.get().add((int) id);
+ } else {
+ LongOpenHashSet overflownHashSet = new LongOpenHashSet((int) INVALID_ID);
+ allLongs.add(overflownHashSet);
+ currentLongs.set(overflownHashSet);
- currentLongs.get().add(id);
- }
+ currentLongs.get().add(id);
}
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/OplogEntryIdSetDrfHashSetThresholdTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/OplogEntryIdSetDrfHashSetThresholdTest.java
new file mode 100644
index 0000000000..ff7e43e7b5
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/OplogEntryIdSetDrfHashSetThresholdTest.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.geode.internal.cache;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.IntStream;
+import java.util.stream.LongStream;
+
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
+import org.junit.jupiter.api.Test;
+import org.junitpioneer.jupiter.SetSystemProperty;
+
+import org.apache.geode.internal.cache.DiskStoreImpl.OplogEntryIdSet;
+
+/**
+ * Tests DiskStoreImpl.OplogEntryIdSet
+ */
+public class OplogEntryIdSetDrfHashSetThresholdTest {
+ @Test
+ @SetSystemProperty(key = "gemfire.disk.drfHashMapOverflowThreshold", value = "10")
+ public void addMethodOverflowBasedOnDrfOverflowThresholdParameters() {
+
+ int testEntries = 41;
+ IntOpenHashSet intOpenHashSet = new IntOpenHashSet();
+ LongOpenHashSet longOpenHashSet = new LongOpenHashSet();
+
+ List<IntOpenHashSet> intOpenHashSets =
+ new ArrayList<>(Collections.singletonList(intOpenHashSet));
+ List<LongOpenHashSet> longOpenHashSets =
+ new ArrayList<>(Collections.singletonList(longOpenHashSet));
+
+ OplogEntryIdSet oplogEntryIdSet = new OplogEntryIdSet(intOpenHashSets, longOpenHashSets);
+ IntStream.range(1, testEntries).forEach(oplogEntryIdSet::add);
+ LongStream.range(0x00000000FFFFFFFFL + 1, 0x00000000FFFFFFFFL + testEntries)
+ .forEach(oplogEntryIdSet::add);
+
+ assertThat(intOpenHashSets).hasSize(4);
+ assertThat(longOpenHashSets).hasSize(4);
+
+ IntStream.range(1, testEntries).forEach(i -> assertThat(oplogEntryIdSet.contains(i)).isTrue());
+ LongStream.range(0x00000000FFFFFFFFL + 1, 0x00000000FFFFFFFFL + testEntries)
+ .forEach(i -> assertThat(oplogEntryIdSet.contains(i)).isTrue());
+
+ }
+}