You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by ag...@apache.org on 2019/03/26 18:44:56 UTC
[geode] branch develop updated: GEODE-6526: Removing call to
removeTombstone during entry destroy (#3348)
This is an automated email from the ASF dual-hosted git repository.
agingade pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new 905e921 GEODE-6526: Removing call to removeTombstone during entry destroy (#3348)
905e921 is described below
commit 905e921169f643e916f56ea391f6891cd1d3ca7c
Author: agingade <ag...@pivotal.io>
AuthorDate: Tue Mar 26 11:44:45 2019 -0700
GEODE-6526: Removing call to removeTombstone during entry destroy (#3348)
* GEODE-6526: Removing call to removeTombstone during entry destroy
During destroy entry (tombstone) under region entry lock, if the
entry had a lower region version than the recorded gc version for
that member, the entry was removed immediately which could result
in dead-lock with tombstone gc thread.
Instead of removing the entry during destroy, it was scheduled to
remove during tombstone gc.
---
.../cache/versions/TombstoneDUnitTest.java | 135 +++++++++++++++++++--
.../geode/internal/cache/InternalRegion.java | 2 +
.../apache/geode/internal/cache/LocalRegion.java | 3 +-
.../cache/entries/AbstractRegionEntry.java | 30 ++---
4 files changed, 141 insertions(+), 29 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 5c8153c..cf4dcb6 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
@@ -17,10 +17,21 @@ package org.apache.geode.internal.cache.versions;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.junit.Assert.assertEquals;
+import java.io.Serializable;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+
import org.junit.Test;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.distributed.internal.ClusterDistributionManager;
+import org.apache.geode.distributed.internal.DistributionMessage;
+import org.apache.geode.distributed.internal.DistributionMessageObserver;
+import org.apache.geode.internal.cache.DestroyOperation;
+import org.apache.geode.internal.cache.DistributedTombstoneOperation;
+import org.apache.geode.internal.cache.LocalRegion;
+import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
@@ -50,7 +61,7 @@ public class TombstoneDUnitTest extends JUnit4CacheTestCase {
Region<String, String> region = getCache().getRegion("TestRegion");
region.destroy("K1");
assertEquals(1, getGemfireCache().getCachePerfStats().getTombstoneCount());
- performGC(region);
+ performGC();
});
vm1.invoke(() -> {
@@ -61,7 +72,7 @@ public class TombstoneDUnitTest extends JUnit4CacheTestCase {
// Send tombstone gc message to vm0.
Region<String, String> region = getCache().getRegion("TestRegion");
region.destroy("K2");
- performGC(region);
+ performGC();
});
vm0.invoke(() -> {
@@ -71,11 +82,118 @@ public class TombstoneDUnitTest extends JUnit4CacheTestCase {
});
}
+ @Test
+ public void testTombstonesWithLowerVersionThanTheRecordedVersionGetsGCed() throws Exception {
+ Host host = Host.getHost(0);
+ VM vm0 = host.getVM(0);
+ VM vm1 = host.getVM(1);
+
+ createCache(vm0);
+ createCache(vm1);
+
+ vm0.invoke(() -> {
+ createRegion("TestRegion", true);
+ Region<String, String> region = getCache().getRegion("TestRegion");
+ region.put("K1", "V1");
+ region.put("K2", "V2");
+ });
+
+ vm1.invoke(() -> {
+ createRegion("TestRegion", false);
+ DistributionMessageObserver.setInstance(new RegionObserver());
+ });
+
+ AsyncInvocation vm0Async1 = vm0.invokeAsync(() -> {
+ Region<String, String> region = getCache().getRegion("TestRegion");
+ region.destroy("K1");
+ });
+
+ AsyncInvocation vm0Async2 = vm0.invokeAsync(() -> {
+ Region<String, String> region = getCache().getRegion("TestRegion");
+ region.destroy("K2");
+ });
+
+ AsyncInvocation vm0Async3 = vm0.invokeAsync(() -> {
+ waitForTombstoneCount(2);
+ performGC(2);
+ });
+
+ vm1.invoke(() -> {
+ await().until(() -> getCache().getCachePerfStats().getTombstoneGCCount() == 1);
+ });
+
+ vm0Async1.join();
+ vm0Async2.join();
+ vm0Async3.join();
+
+ vm1.invoke(() -> {
+ Region<String, String> region = getCache().getRegion("TestRegion");
+ performGC(((LocalRegion) region).getTombstoneCount());
+ assertEquals(0, ((LocalRegion) region).getTombstoneCount());
+ });
+ }
+
+ private class RegionObserver extends DistributionMessageObserver implements Serializable {
+
+ VersionTag versionTag = null;
+ CountDownLatch tombstoneGcLatch = new CountDownLatch(1);
+
+ @Override
+ public void beforeProcessMessage(ClusterDistributionManager dm, DistributionMessage message) {
+ // Allow destroy with higher version to complete first.
+ if (message instanceof DestroyOperation.DestroyMessage) {
+ // wait for tombstoneGC message to complete.
+ try {
+ tombstoneGcLatch.await();
+ synchronized (this) {
+ DestroyOperation.DestroyMessage destroyMessage =
+ (DestroyOperation.DestroyMessage) message;
+ if (versionTag == null) {
+ // First destroy
+ versionTag = destroyMessage.getVersionTag();
+ this.wait();
+ } else {
+ // Second destroy
+ if (destroyMessage.getVersionTag().getRegionVersion() < versionTag
+ .getRegionVersion()) {
+ this.notifyAll();
+ this.wait();
+ }
+ }
+ }
+ } catch (InterruptedException ex) {
+ }
+ }
+ }
+
+ @Override
+ public void afterProcessMessage(ClusterDistributionManager dm, DistributionMessage message) {
+ if (message instanceof DestroyOperation.DestroyMessage) {
+ // Notify the destroy with smaller version to continue.
+ synchronized (this) {
+ this.notifyAll();
+ }
+ }
+ if (message instanceof DistributedTombstoneOperation.TombstoneMessage) {
+ tombstoneGcLatch.countDown();
+ }
+ }
+ };
+
+ private void createCache(VM vm) {
+ vm.invoke(() -> {
+ if (cache != null && !cache.isClosed()) {
+ cache.close();
+ }
+ Properties props = new Properties();
+ props.put("conserve-sockets", "false");
+ cache = getCache(props);
+ });
+ }
+
private void waitForTombstoneCount(int count) {
try {
- await().until(() -> {
- return getGemfireCache().getCachePerfStats().getTombstoneCount() == count;
- });
+ await().until(() -> getCache().getCachePerfStats().getTombstoneCount() == count);
} catch (Exception e) {
// The caller to throw exception with proper message.
}
@@ -89,8 +207,11 @@ public class TombstoneDUnitTest extends JUnit4CacheTestCase {
}
}
- private void performGC(Region region) throws Exception {
- getGemfireCache().getTombstoneService().forceBatchExpirationForTests(1);
+ private void performGC(int count) throws Exception {
+ getCache().getTombstoneService().forceBatchExpirationForTests(count);
}
+ private void performGC() throws Exception {
+ performGC(1);
+ }
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalRegion.java
index 509cede..f6da64c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalRegion.java
@@ -102,6 +102,8 @@ public interface InternalRegion extends Region, HasCachePerfStats, RegionEntryCo
void scheduleTombstone(RegionEntry entry, VersionTag destroyedVersion);
+ void scheduleTombstone(RegionEntry entry, VersionTag destroyedVersion, boolean reschedule);
+
boolean isEntryExpiryPossible();
void addExpiryTaskIfAbsent(RegionEntry entry);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
index 35b0630..7e40199 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
@@ -3265,7 +3265,8 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
scheduleTombstone(entry, destroyedVersion, false);
}
- private void scheduleTombstone(RegionEntry entry, VersionTag destroyedVersion,
+ @Override
+ public void scheduleTombstone(RegionEntry entry, VersionTag destroyedVersion,
boolean reschedule) {
if (destroyedVersion == null) {
throw new NullPointerException("destroyed version tag cannot be null");
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/entries/AbstractRegionEntry.java b/geode-core/src/main/java/org/apache/geode/internal/cache/entries/AbstractRegionEntry.java
index 191d9d9..b044fc3 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/entries/AbstractRegionEntry.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/entries/AbstractRegionEntry.java
@@ -257,27 +257,15 @@ public abstract class AbstractRegionEntry implements HashRegionEntry<Object, Obj
throws RegionClearedException {
assert region.getVersionVector() != null;
assert version != null;
- if (region.getServerProxy() == null && region.getVersionVector()
- .isTombstoneTooOld(version.getMemberID(), version.getRegionVersion())) {
- // distributed gc with higher vector version preempts this operation
- if (!isTombstone()) {
- basicMakeTombstone(region);
- region.getCachePerfStats().incTombstoneCount(1);
- }
- ((DiskRecoveryStore) region).getRegionMap().removeTombstone(this, version, false, true);
- } else {
- if (isTombstone()) {
- // unschedule the old tombstone
- region.unscheduleTombstone(this);
- }
- setRecentlyUsed(region);
- boolean newEntry = getValueAsToken() == Token.REMOVED_PHASE1;
- basicMakeTombstone(region);
- region.scheduleTombstone(this, version);
- if (newEntry) {
- // bug #46631 - entry count is decremented by scheduleTombstone but this is a new entry
- region.getCachePerfStats().incEntryCount(1);
- }
+
+ boolean wasTombstone = isTombstone();
+ setRecentlyUsed(region);
+ boolean newEntry = getValueAsToken() == Token.REMOVED_PHASE1;
+ basicMakeTombstone(region);
+ region.scheduleTombstone(this, version, wasTombstone);
+ if (newEntry) {
+ // bug #46631 - entry count is decremented by scheduleTombstone but this is a new entry
+ region.getCachePerfStats().incEntryCount(1);
}
}