You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by zh...@apache.org on 2020/07/09 22:57:23 UTC
[geode] branch feature/GEODE-8334 created (now 43cb278)
This is an automated email from the ASF dual-hosted git repository.
zhouxj pushed a change to branch feature/GEODE-8334
in repository https://gitbox.apache.org/repos/asf/geode.git.
at 43cb278 GEODE-8334: PR.clear should sync with putAll or removeAll on rvvLock
This branch includes the following new commits:
new 43cb278 GEODE-8334: PR.clear should sync with putAll or removeAll on rvvLock
The 1 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.
[geode] 01/01: GEODE-8334: PR.clear should sync with putAll or
removeAll on rvvLock
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhouxj pushed a commit to branch feature/GEODE-8334
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 43cb2781e34b0b24f2628236256e13027f4b4e50
Author: zhouxh <gz...@pivotal.io>
AuthorDate: Thu Jul 9 15:55:15 2020 -0700
GEODE-8334: PR.clear should sync with putAll or removeAll on rvvLock
Co-authored-by: Xiaojian Zhou <gz...@pivotal.io>
Co-authored-by: Anil <ag...@pivotal.io>
---
.../apache/geode/internal/cache/BucketRegion.java | 8 +----
.../cache/partitioned/PutAllPRMessage.java | 9 ++++++
.../cache/partitioned/RemoveAllPRMessage.java | 9 ++++++
.../cache/partitioned/PutAllPRMessageTest.java | 33 +++++++++++++++++++++
.../cache/partitioned/RemoveAllPRMessageTest.java | 34 ++++++++++++++++++++++
5 files changed, 86 insertions(+), 7 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
index c37e1a3..6793ecb 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
@@ -566,12 +566,6 @@ public class BucketRegion extends DistributedRegion implements Bucket {
return;
}
- boolean enableRVV = useRVV && getConcurrencyChecksEnabled();
- RegionVersionVector rvv = null;
- if (enableRVV) {
- rvv = getVersionVector().getCloneForTransmission();
- }
-
// get rvvLock
Set<InternalDistributedMember> participants =
getCacheDistributionAdvisor().adviseInvalidateRegion();
@@ -585,7 +579,7 @@ public class BucketRegion extends DistributedRegion implements Bucket {
// no need to dominate my own rvv.
// Clear is on going here, there won't be GII for this member
clearRegionLocally(regionEvent, cacheWrite, null);
- distributeClearOperation(regionEvent, rvv, participants);
+ distributeClearOperation(regionEvent, null, participants);
// TODO: call reindexUserDataRegion if there're lucene indexes
} finally {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java
index 0c690c5..6bb666c 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java
@@ -413,6 +413,7 @@ public class PutAllPRMessage extends PartitionMessageWithDirectReply {
Object[] keys = getKeysToBeLocked();
if (!notificationOnly) {
boolean locked = false;
+ boolean rvvLocked = false;
try {
if (putAllPRData.length > 0) {
if (this.posDup && bucketRegion.getConcurrencyChecksEnabled()) {
@@ -438,6 +439,10 @@ public class PutAllPRMessage extends PartitionMessageWithDirectReply {
bucketRegion.recordBulkOpStart(membershipID, eventID);
}
locked = bucketRegion.waitUntilLocked(keys);
+ if (!rvvLocked) {
+ bucketRegion.lockRVVForBulkOp();
+ rvvLocked = true;
+ }
boolean lockedForPrimary = false;
final HashMap succeeded = new HashMap();
PutAllPartialResult partialKeys = new PutAllPartialResult(putAllPRDataSize);
@@ -518,6 +523,10 @@ public class PutAllPRMessage extends PartitionMessageWithDirectReply {
} catch (RegionDestroyedException e) {
ds.checkRegionDestroyedOnBucket(bucketRegion, true, e);
} finally {
+ if (rvvLocked) {
+ bucketRegion.unlockRVVForBulkOp();
+ rvvLocked = false;
+ }
if (locked) {
bucketRegion.removeAndNotifyKeys(keys);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/RemoveAllPRMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/RemoveAllPRMessage.java
index 6f355d6..f295136 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/RemoveAllPRMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/RemoveAllPRMessage.java
@@ -406,6 +406,7 @@ public class RemoveAllPRMessage extends PartitionMessageWithDirectReply {
if (!notificationOnly) {
boolean locked = false;
+ boolean rvvLocked = false;
try {
if (removeAllPRData.length > 0) {
if (this.posDup && bucketRegion.getConcurrencyChecksEnabled()) {
@@ -431,6 +432,10 @@ public class RemoveAllPRMessage extends PartitionMessageWithDirectReply {
bucketRegion.recordBulkOpStart(membershipID, eventID);
}
locked = bucketRegion.waitUntilLocked(keys);
+ if (!rvvLocked) {
+ bucketRegion.lockRVVForBulkOp();
+ rvvLocked = true;
+ }
boolean lockedForPrimary = false;
final ArrayList<Object> succeeded = new ArrayList<Object>();
PutAllPartialResult partialKeys = new PutAllPartialResult(removeAllPRDataSize);
@@ -526,6 +531,10 @@ public class RemoveAllPRMessage extends PartitionMessageWithDirectReply {
} catch (RegionDestroyedException e) {
ds.checkRegionDestroyedOnBucket(bucketRegion, true, e);
} finally {
+ if (rvvLocked) {
+ bucketRegion.unlockRVVForBulkOp();
+ rvvLocked = false;
+ }
if (locked) {
bucketRegion.removeAndNotifyKeys(keys);
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessageTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessageTest.java
index ab82a93..c5dd140 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessageTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessageTest.java
@@ -15,9 +15,12 @@
package org.apache.geode.internal.cache.partitioned;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
@@ -119,4 +122,34 @@ public class PutAllPRMessageTest {
eq(regionDestroyedException));
}
+ @Test
+ public void rvvLockedAfterKeysAreLockedAndUnlockRVVBeforeKeys() throws Exception {
+ PutAllPRMessage message = spy(new PutAllPRMessage(bucketId, 1, false, false, false, null));
+ message.addEntry(entryData);
+ doReturn(keys).when(message).getKeysToBeLocked();
+ when(bucketRegion.waitUntilLocked(keys)).thenReturn(true);
+ when(bucketRegion.doLockForPrimary(false)).thenThrow(new PrimaryBucketException());
+ doNothing().when(bucketRegion).lockRVVForBulkOp();
+ doNothing().when(bucketRegion).unlockRVVForBulkOp();
+
+ InternalCache cache = mock(InternalCache.class);
+ InternalDistributedSystem ids = mock(InternalDistributedSystem.class);
+ when(bucketRegion.getCache()).thenReturn(cache);
+ when(cache.getDistributedSystem()).thenReturn(ids);
+ when(ids.getOffHeapStore()).thenReturn(null);
+
+ try {
+ message.doLocalPutAll(partitionedRegion, mock(InternalDistributedMember.class), 1);
+ fail("Expect PrimaryBucketException");
+ } catch (Exception e) {
+ assertThat(e instanceof PrimaryBucketException);
+ }
+
+ InOrder inOrder = inOrder(bucketRegion);
+ inOrder.verify(bucketRegion).waitUntilLocked(keys);
+ inOrder.verify(bucketRegion).lockRVVForBulkOp();
+ inOrder.verify(bucketRegion).unlockRVVForBulkOp();
+ inOrder.verify(bucketRegion).removeAndNotifyKeys(keys);
+ }
+
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/partitioned/RemoveAllPRMessageTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/partitioned/RemoveAllPRMessageTest.java
index 2309cb0..19d508e 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/partitioned/RemoveAllPRMessageTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/partitioned/RemoveAllPRMessageTest.java
@@ -14,9 +14,12 @@
*/
package org.apache.geode.internal.cache.partitioned;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
@@ -131,4 +134,35 @@ public class RemoveAllPRMessageTest {
verify(dataStore).checkRegionDestroyedOnBucket(eq(bucketRegion), eq(true),
eq(regionDestroyedException));
}
+
+ @Test
+ public void rvvLockedAfterKeysAreLockedAndUnlockRVVBeforeKeys() throws Exception {
+ RemoveAllPRMessage message =
+ spy(new RemoveAllPRMessage(bucketId, 1, false, false, false, null));
+ message.addEntry(entryData);
+ doReturn(keys).when(message).getKeysToBeLocked();
+ when(bucketRegion.waitUntilLocked(keys)).thenReturn(true);
+ when(bucketRegion.doLockForPrimary(false)).thenThrow(new PrimaryBucketException());
+ doNothing().when(bucketRegion).lockRVVForBulkOp();
+ doNothing().when(bucketRegion).unlockRVVForBulkOp();
+
+ InternalCache cache = mock(InternalCache.class);
+ InternalDistributedSystem ids = mock(InternalDistributedSystem.class);
+ when(bucketRegion.getCache()).thenReturn(cache);
+ when(cache.getDistributedSystem()).thenReturn(ids);
+ when(ids.getOffHeapStore()).thenReturn(null);
+
+ try {
+ message.doLocalRemoveAll(partitionedRegion, mock(InternalDistributedMember.class), true);
+ fail("Expect PrimaryBucketException");
+ } catch (Exception e) {
+ assertThat(e instanceof PrimaryBucketException);
+ }
+
+ InOrder inOrder = inOrder(bucketRegion);
+ inOrder.verify(bucketRegion).waitUntilLocked(keys);
+ inOrder.verify(bucketRegion).lockRVVForBulkOp();
+ inOrder.verify(bucketRegion).unlockRVVForBulkOp();
+ inOrder.verify(bucketRegion).removeAndNotifyKeys(keys);
+ }
}