You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bs...@apache.org on 2019/01/18 00:56:45 UTC
[geode] branch feature/GEODE-2113e updated:
SerialGatewaySenderDistributedDeadlockDUnitTest changes for stresstest
This is an automated email from the ASF dual-hosted git repository.
bschuchardt pushed a commit to branch feature/GEODE-2113e
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/feature/GEODE-2113e by this push:
new 91db36a SerialGatewaySenderDistributedDeadlockDUnitTest changes for stresstest
91db36a is described below
commit 91db36a69e0e9493d464a3b66fddee707ccc8884
Author: Bruce Schuchardt <bs...@pivotal.io>
AuthorDate: Thu Jan 17 13:47:29 2019 -0800
SerialGatewaySenderDistributedDeadlockDUnitTest changes for stresstest
This test failed in code modified on this branch. I've added an
assertion in MsgReader and modified the test to add names to the >120
lambdas it uses so I can make sense of what's going on.
---
.../geode/ClusterCommunicationsDUnitTest.java | 2 +-
.../membership/gms/membership/GMSJoinLeave.java | 2 +-
.../apache/geode/internal/net/NioPlainEngine.java | 4 +-
.../org/apache/geode/internal/tcp/MsgReader.java | 2 +
.../org/apache/geode/internal/net/BuffersTest.java | 2 -
.../geode/internal/net/NioPlainEngineTest.java | 154 ++++++++++++++
.../geode/internal/net/NioSslEngineTest.java | 3 +-
.../geode/internal/cache/wan/WANTestBase.java | 1 -
...lGatewaySenderDistributedDeadlockDUnitTest.java | 229 +++++++++++----------
9 files changed, 286 insertions(+), 113 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/ClusterCommunicationsDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/ClusterCommunicationsDUnitTest.java
index d5c30c4..6a1e0a5 100644
--- a/geode-core/src/distributedTest/java/org/apache/geode/ClusterCommunicationsDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/ClusterCommunicationsDUnitTest.java
@@ -184,7 +184,7 @@ public class ClusterCommunicationsDUnitTest implements java.io.Serializable {
VM.getVM(2).invoke(() -> cache.getDistributedSystem().getDistributedMember());
VM.getVM(1).invoke("receive a large direct-reply message", () -> {
SerialAckedMessageWithBigReply messageWithBigReply = new SerialAckedMessageWithBigReply();
- await().atMost(30, TimeUnit.SECONDS).until(() -> {
+ await().atMost(3, TimeUnit.MINUTES).until(() -> {
messageWithBigReply.send(Collections.<DistributedMember>singleton(vm2ID));
return true;
});
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeave.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeave.java
index e72c75f..01ea87c 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeave.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeave.java
@@ -1152,7 +1152,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
}
}
} catch (IOException | ClassNotFoundException problem) {
- logger.debug("EOFException IOException ", problem);
+ logger.debug("Exception caught while trying to join", problem);
}
}
} while (!anyResponses && System.currentTimeMillis() < giveUpTime);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/net/NioPlainEngine.java b/geode-core/src/main/java/org/apache/geode/internal/net/NioPlainEngine.java
index 8e49722..85fab7f 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/net/NioPlainEngine.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/net/NioPlainEngine.java
@@ -32,8 +32,8 @@ import org.apache.geode.internal.logging.LogService;
public class NioPlainEngine implements NioFilter {
private static final Logger logger = LogService.getLogger();
- private int lastReadPosition;
- private int lastProcessedPosition;
+ int lastReadPosition;
+ int lastProcessedPosition;
public NioPlainEngine() {}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java b/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java
index 0fc96ad..6470203 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java
@@ -20,6 +20,7 @@ import java.nio.ByteBuffer;
import org.apache.geode.distributed.internal.DMStats;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.ReplyProcessor21;
+import org.apache.geode.internal.Assert;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.Version;
import org.apache.geode.internal.net.Buffers;
@@ -75,6 +76,7 @@ public class MsgReader {
DistributionMessage readMessage(Header header)
throws IOException, ClassNotFoundException {
ByteBuffer nioInputBuffer = readAtLeast(header.messageLength);
+ Assert.assertTrue(nioInputBuffer.remaining() >= header.messageLength);
this.getStats().incMessagesBeingReceived(true, header.messageLength);
long startSer = this.getStats().startMsgDeserialization();
try {
diff --git a/geode-core/src/test/java/org/apache/geode/internal/net/BuffersTest.java b/geode-core/src/test/java/org/apache/geode/internal/net/BuffersTest.java
index 8355e62..65f8581 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/net/BuffersTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/net/BuffersTest.java
@@ -38,8 +38,6 @@ public class BuffersTest {
createAndVerifyNewWriteBuffer(buffer, true);
- // buffer.limit(buffer.capacity());
- // buffer.position(0);
createAndVerifyNewReadBuffer(buffer, false);
diff --git a/geode-core/src/test/java/org/apache/geode/internal/net/NioPlainEngineTest.java b/geode-core/src/test/java/org/apache/geode/internal/net/NioPlainEngineTest.java
new file mode 100644
index 0000000..a7eb2d7
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/net/NioPlainEngineTest.java
@@ -0,0 +1,154 @@
+/*
+ * 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.net;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.EOFException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import org.apache.geode.distributed.internal.DMStats;
+
+public class NioPlainEngineTest {
+
+ private static final int netBufferSize = 10000;
+ private static final int appBufferSize = 20000;
+
+ private DMStats mockStats;
+ private NioPlainEngine nioEngine;
+
+ @Before
+ public void setUp() throws Exception {
+ mockStats = mock(DMStats.class);
+
+ nioEngine = new NioPlainEngine();
+ }
+
+ @Test
+ public void unwrap() {
+ ByteBuffer buffer = ByteBuffer.allocate(100);
+ buffer.position(0).limit(buffer.capacity());
+ nioEngine.unwrap(buffer);
+ assertThat(buffer.position()).isEqualTo(buffer.limit());
+ }
+
+ @Test
+ public void ensureWrappedCapacity() {
+ ByteBuffer wrappedBuffer = ByteBuffer.allocate(100);
+ wrappedBuffer.put(new byte[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
+ nioEngine.lastReadPosition = 10;
+ int requestedCapacity = 210;
+ ByteBuffer result = nioEngine.ensureWrappedCapacity(requestedCapacity, wrappedBuffer,
+ Buffers.BufferType.UNTRACKED, mockStats);
+ assertThat(result.capacity()).isGreaterThanOrEqualTo(requestedCapacity);
+ assertThat(result).isNotSameAs(wrappedBuffer);
+ // make sure that data was transferred to the new buffer
+ for (int i = 0; i < 10; i++) {
+ assertThat(result.get(i)).isEqualTo(wrappedBuffer.get(i));
+ }
+ }
+
+ @Test
+ public void ensureWrappedCapacityWithEnoughExistingCapacityAndConsumedDataPresent() {
+ int requestedCapacity = 210;
+ final int consumedDataPresentInBuffer = 100;
+ final int unconsumedDataPresentInBuffer = 10;
+ // the buffer will have enough capacity but will need to be compacted
+ ByteBuffer wrappedBuffer =
+ ByteBuffer.allocate(requestedCapacity + unconsumedDataPresentInBuffer);
+ wrappedBuffer.put(new byte[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
+ nioEngine.lastProcessedPosition = consumedDataPresentInBuffer;
+ // previous read left 10 bytes
+ nioEngine.lastReadPosition = consumedDataPresentInBuffer + unconsumedDataPresentInBuffer;
+ ByteBuffer result =
+ wrappedBuffer = nioEngine.ensureWrappedCapacity(requestedCapacity, wrappedBuffer,
+ Buffers.BufferType.UNTRACKED, mockStats);
+ assertThat(result.capacity()).isEqualTo(requestedCapacity + unconsumedDataPresentInBuffer);
+ assertThat(result).isSameAs(wrappedBuffer);
+ // make sure that data was transferred to the new buffer
+ for (int i = 0; i < 10; i++) {
+ assertThat(result.get(i)).isEqualTo(wrappedBuffer.get(i));
+ }
+ assertThat(nioEngine.lastProcessedPosition).isEqualTo(0);
+ assertThat(nioEngine.lastReadPosition).isEqualTo(10);
+ }
+
+ @Test
+ public void readAtLeast() throws Exception {
+ final int amountToRead = 150;
+ final int individualRead = 60;
+ final int preexistingBytes = 10;
+ ByteBuffer wrappedBuffer = ByteBuffer.allocate(1000);
+ SocketChannel mockChannel = mock(SocketChannel.class);
+
+ // simulate some socket reads
+ when(mockChannel.read(any(ByteBuffer.class))).thenAnswer(new Answer<Integer>() {
+ @Override
+ public Integer answer(InvocationOnMock invocation) throws Throwable {
+ ByteBuffer buffer = invocation.getArgument(0);
+ buffer.position(buffer.position() + individualRead);
+ return individualRead;
+ }
+ });
+
+ nioEngine.lastReadPosition = 10;
+
+ ByteBuffer data = nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer, mockStats);
+ verify(mockChannel, times(3)).read(isA(ByteBuffer.class));
+ assertThat(data.position()).isEqualTo(0);
+ assertThat(data.limit()).isEqualTo(amountToRead);
+ assertThat(nioEngine.lastReadPosition).isEqualTo(individualRead * 3 + preexistingBytes);
+ assertThat(nioEngine.lastProcessedPosition).isEqualTo(amountToRead);
+
+ data = nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer, mockStats);
+ verify(mockChannel, times(5)).read(any(ByteBuffer.class));
+ // at end of last readAtLeast data
+ assertThat(data.position()).isEqualTo(amountToRead);
+ // we read amountToRead bytes
+ assertThat(data.limit()).isEqualTo(amountToRead * 2);
+ // we did 2 more reads from the network
+ assertThat(nioEngine.lastReadPosition).isEqualTo(individualRead * 5 + preexistingBytes);
+ // the next read will start at the end of consumed data
+ assertThat(nioEngine.lastProcessedPosition).isEqualTo(amountToRead * 2);
+
+ }
+
+ @Test(expected = EOFException.class)
+ public void readAtLeastThrowsEOFException() throws Exception {
+ final int amountToRead = 150;
+ ByteBuffer wrappedBuffer = ByteBuffer.allocate(1000);
+ SocketChannel mockChannel = mock(SocketChannel.class);
+
+ // simulate some socket reads
+ when(mockChannel.read(any(ByteBuffer.class))).thenReturn(-1);
+
+ nioEngine.lastReadPosition = 10;
+
+ nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer, mockStats);
+ }
+
+}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java b/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java
index 63348f8..f10c45d 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java
@@ -66,7 +66,6 @@ public class NioSslEngineTest {
private SSLEngine mockEngine;
private DMStats mockStats;
- private SSLSession mockSession;
private NioSslEngine nioSslEngine;
private NioSslEngine spyNioSslEngine;
@@ -74,7 +73,7 @@ public class NioSslEngineTest {
public void setUp() throws Exception {
mockEngine = mock(SSLEngine.class);
- mockSession = mock(SSLSession.class);
+ SSLSession mockSession = mock(SSLSession.class);
when(mockEngine.getSession()).thenReturn(mockSession);
when(mockSession.getPacketBufferSize()).thenReturn(netBufferSize);
when(mockSession.getApplicationBufferSize()).thenReturn(appBufferSize);
diff --git a/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
index 035d55a..70fbe1c 100644
--- a/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
+++ b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
@@ -2816,7 +2816,6 @@ public class WANTestBase extends DistributedTestCase {
});
for (int i = 0; i < regionSize; i++) {
final int temp = i;
- logger.info("For Key : Key_" + i + " : Values : " + r.get("Key_" + i));
await()
.untilAsserted(() -> assertEquals(
"keySet = " + r.keySet() + " values() = " + r.values() + "Region Size = " + r.size(),
diff --git a/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderDistributedDeadlockDUnitTest.java b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderDistributedDeadlockDUnitTest.java
index 0394fb0..40e984b 100644
--- a/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderDistributedDeadlockDUnitTest.java
+++ b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderDistributedDeadlockDUnitTest.java
@@ -30,7 +30,6 @@ import org.apache.geode.cache.execute.RegionFunctionContext;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.internal.cache.wan.WANTestBase;
import org.apache.geode.test.dunit.AsyncInvocation;
-import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.junit.categories.WanTest;
// The tests here are to validate changes introduced because a distributed deadlock
@@ -70,13 +69,13 @@ public class SerialGatewaySenderDistributedDeadlockDUnitTest extends WANTestBase
// exercise region and gateway operations with different messaging
exerciseWANOperations();
AsyncInvocation invVM4transaction =
- vm4.invokeAsync(() -> WANTestBase.doTxPuts(getTestMethodName() + "_RR"));
+ vm4.invokeAsync("doTxPutsAsync", () -> WANTestBase.doTxPuts(getTestMethodName() + "_RR"));
AsyncInvocation invVM5transaction =
- vm5.invokeAsync(() -> WANTestBase.doTxPuts(getTestMethodName() + "_RR"));
+ vm5.invokeAsync("doTxPutsAsync", () -> WANTestBase.doTxPuts(getTestMethodName() + "_RR"));
AsyncInvocation invVM4 =
- vm4.invokeAsync(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
+ vm4.invokeAsync("doPutsAsync", () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
AsyncInvocation invVM5 =
- vm5.invokeAsync(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
+ vm5.invokeAsync("doPutsAsync", () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
exerciseFunctions();
@@ -95,9 +94,11 @@ public class SerialGatewaySenderDistributedDeadlockDUnitTest extends WANTestBase
// Uses partitioned regions and conserve-sockets=false
@Test
public void testPrimarySendersOnDifferentVMsPR() throws Exception {
- Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstPeerLocator(1));
+ Integer lnPort =
+ (Integer) vm0.invoke("createFirstPeerLocator", () -> WANTestBase.createFirstPeerLocator(1));
- Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+ Integer nyPort = (Integer) vm1.invoke("createFirstRemoteLocator",
+ () -> WANTestBase.createFirstRemoteLocator(2, lnPort));
createCachesWith(Boolean.FALSE, nyPort, lnPort);
@@ -109,37 +110,34 @@ public class SerialGatewaySenderDistributedDeadlockDUnitTest extends WANTestBase
exerciseWANOperations();
AsyncInvocation invVM4transaction =
- vm4.invokeAsync(() -> SerialGatewaySenderDistributedDeadlockDUnitTest
+ vm4.invokeAsync("doTxPutsPRAsync", () -> SerialGatewaySenderDistributedDeadlockDUnitTest
.doTxPutsPR(getTestMethodName() + "_RR", 100, 1000));
AsyncInvocation invVM5transaction =
- vm5.invokeAsync(() -> SerialGatewaySenderDistributedDeadlockDUnitTest
+ vm5.invokeAsync("doTxPutsPRAsync", () -> SerialGatewaySenderDistributedDeadlockDUnitTest
.doTxPutsPR(getTestMethodName() + "_RR", 100, 1000));
AsyncInvocation invVM4 =
- vm4.invokeAsync(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
+ vm4.invokeAsync("doPutsAsync", () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
AsyncInvocation invVM5 =
- vm5.invokeAsync(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
+ vm5.invokeAsync("doPutsAsync", () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
exerciseFunctions();
- try {
- invVM4transaction.join();
- invVM5transaction.join();
- invVM4.join();
- invVM5.join();
- } catch (InterruptedException e) {
- e.printStackTrace();
- fail();
- }
+ invVM4transaction.join();
+ invVM5transaction.join();
+ invVM4.join();
+ invVM5.join();
}
// Uses replicated regions and conserve-sockets=true
@Test
public void testPrimarySendersOnDifferentVMsReplicatedSocketPolicy() throws Exception {
- Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstPeerLocator(1));
+ Integer lnPort =
+ (Integer) vm0.invoke("createFirstPeerLocator", () -> WANTestBase.createFirstPeerLocator(1));
- Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+ Integer nyPort = (Integer) vm1.invoke("createFirstRemoteLocator",
+ () -> WANTestBase.createFirstRemoteLocator(2, lnPort));
createCachesWith(Boolean.TRUE, nyPort, lnPort);
@@ -154,35 +152,32 @@ public class SerialGatewaySenderDistributedDeadlockDUnitTest extends WANTestBase
// exercise region and gateway operations with messaging
exerciseWANOperations();
AsyncInvocation invVM4transaction =
- vm4.invokeAsync(() -> WANTestBase.doTxPuts(getTestMethodName() + "_RR"));
+ vm4.invokeAsync("doTxPutsAsync", () -> WANTestBase.doTxPuts(getTestMethodName() + "_RR"));
AsyncInvocation invVM5transaction =
- vm5.invokeAsync(() -> WANTestBase.doTxPuts(getTestMethodName() + "_RR"));
+ vm5.invokeAsync("doTxPutsAsync", () -> WANTestBase.doTxPuts(getTestMethodName() + "_RR"));
AsyncInvocation invVM4 =
- vm4.invokeAsync(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
+ vm4.invokeAsync("doPutsAsync", () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
AsyncInvocation invVM5 =
- vm5.invokeAsync(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
+ vm5.invokeAsync("doPutsAsync", () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
exerciseFunctions();
- try {
- invVM4transaction.join();
- invVM5transaction.join();
- invVM4.join();
- invVM5.join();
- } catch (InterruptedException e) {
- e.printStackTrace();
- fail();
- }
+ invVM4transaction.join();
+ invVM5transaction.join();
+ invVM4.join();
+ invVM5.join();
}
// Uses partitioned regions and conserve-sockets=true
// this always causes a distributed deadlock
@Test
public void testPrimarySendersOnDifferentVMsPRSocketPolicy() throws Exception {
- Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstPeerLocator(1));
+ Integer lnPort =
+ (Integer) vm0.invoke("createFirstPeerLocator", () -> WANTestBase.createFirstPeerLocator(1));
- Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+ Integer nyPort = (Integer) vm1.invoke("createFirstRemoteLocator",
+ () -> WANTestBase.createFirstRemoteLocator(2, lnPort));
createCachesWith(Boolean.TRUE, nyPort, lnPort);
@@ -194,28 +189,23 @@ public class SerialGatewaySenderDistributedDeadlockDUnitTest extends WANTestBase
exerciseWANOperations();
AsyncInvocation invVM4transaction =
- vm4.invokeAsync(() -> SerialGatewaySenderDistributedDeadlockDUnitTest
+ vm4.invokeAsync("doTxPutsPRAsync", () -> SerialGatewaySenderDistributedDeadlockDUnitTest
.doTxPutsPR(getTestMethodName() + "_RR", 100, 1000));
AsyncInvocation invVM5transaction =
- vm5.invokeAsync(() -> SerialGatewaySenderDistributedDeadlockDUnitTest
+ vm5.invokeAsync("doTxPutsPRAsync", () -> SerialGatewaySenderDistributedDeadlockDUnitTest
.doTxPutsPR(getTestMethodName() + "_RR", 100, 1000));
AsyncInvocation invVM4 =
- vm4.invokeAsync(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
+ vm4.invokeAsync("doPutsAsync", () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
AsyncInvocation invVM5 =
- vm5.invokeAsync(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
+ vm5.invokeAsync("doPutsAsync", () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 1000));
exerciseFunctions();
- try {
- invVM4transaction.join();
- invVM5transaction.join();
- invVM4.join();
- invVM5.join();
- } catch (InterruptedException e) {
- e.printStackTrace();
- fail();
- }
+ invVM4transaction.join();
+ invVM5transaction.join();
+ invVM4.join();
+ invVM5.join();
}
// **************************************************************************
@@ -223,58 +213,64 @@ public class SerialGatewaySenderDistributedDeadlockDUnitTest extends WANTestBase
// **************************************************************************
private void createReplicatedRegions(Integer nyPort) throws Exception {
// create receiver
- vm2.invoke(() -> WANTestBase.createReplicatedRegion(getTestMethodName() + "_RR", null, false));
- vm2.invoke(() -> WANTestBase.createReceiver());
+ vm2.invoke("createReplicatedRegion",
+ () -> WANTestBase.createReplicatedRegion(getTestMethodName() + "_RR", null, false));
+ vm2.invoke("createReceiver", () -> WANTestBase.createReceiver());
// create senders
- vm4.invoke(
+ vm4.invoke("createReplicatedRegion",
() -> WANTestBase.createReplicatedRegion(getTestMethodName() + "_RR", "ln1,ln2", false));
- vm5.invoke(
+ vm5.invoke("createReplicatedRegion",
() -> WANTestBase.createReplicatedRegion(getTestMethodName() + "_RR", "ln1,ln2", false));
}
private void createCachesWith(Boolean socketPolicy, Integer nyPort, Integer lnPort) {
- vm2.invoke(() -> WANTestBase.createCacheConserveSockets(socketPolicy, nyPort));
+ vm2.invoke("createCacheConserveSockets",
+ () -> WANTestBase.createCacheConserveSockets(socketPolicy, nyPort));
- vm4.invoke(() -> WANTestBase.createCacheConserveSockets(socketPolicy, lnPort));
+ vm4.invoke("createCacheConserveSockets",
+ () -> WANTestBase.createCacheConserveSockets(socketPolicy, lnPort));
- vm5.invoke(() -> WANTestBase.createCacheConserveSockets(socketPolicy, lnPort));
+ vm5.invoke("createCacheConserveSockets",
+ () -> WANTestBase.createCacheConserveSockets(socketPolicy, lnPort));
}
private void exerciseFunctions() throws Exception {
// do function calls that use a shared connection
for (int x = 0; x < 1000; x++) {
// setting it to Boolean.TRUE it should pass the test
- vm4.invoke(() -> SerialGatewaySenderDistributedDeadlockDUnitTest
+ vm4.invoke("doFunctionPuts", () -> SerialGatewaySenderDistributedDeadlockDUnitTest
.doFunctionPuts(getTestMethodName() + "_RR", 1, Boolean.TRUE));
- vm5.invoke(() -> SerialGatewaySenderDistributedDeadlockDUnitTest
+ vm5.invoke("doFunctionPuts", () -> SerialGatewaySenderDistributedDeadlockDUnitTest
.doFunctionPuts(getTestMethodName() + "_RR", 1, Boolean.TRUE));
}
for (int x = 0; x < 1000; x++) {
// setting the Boolean.FALSE below will cause a deadlock in some GFE versions
// setting it to Boolean.TRUE as above it should pass the test
// this is similar to the customer found distributed deadlock
- vm4.invoke(() -> SerialGatewaySenderDistributedDeadlockDUnitTest
+ vm4.invoke("doFunctionPuts2", () -> SerialGatewaySenderDistributedDeadlockDUnitTest
.doFunctionPuts(getTestMethodName() + "_RR", 1, Boolean.FALSE));
- vm5.invoke(() -> SerialGatewaySenderDistributedDeadlockDUnitTest
+ vm5.invoke("doFunctionPuts2", () -> SerialGatewaySenderDistributedDeadlockDUnitTest
.doFunctionPuts(getTestMethodName() + "_RR", 1, Boolean.FALSE));
}
}
private void createPartitionedRegions(Integer nyPort) throws Exception {
// create remote receiver
- vm2.invoke(
+ vm2.invoke("createPartitionedRegion",
() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_RR", "", 0, 113, false));
- vm2.invoke(() -> WANTestBase.createReceiver());
+ vm2.invoke("createReceiver", () -> WANTestBase.createReceiver());
// create sender vms
- vm4.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_RR", "ln1,ln2", 1,
- 113, false));
+ vm4.invoke("createPartitionedRegion",
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_RR", "ln1,ln2", 1,
+ 113, false));
- vm5.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_RR", "ln1,ln2", 1,
- 113, false));
+ vm5.invoke("createPartitionedRegion",
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_RR", "ln1,ln2", 1,
+ 113, false));
}
private void exerciseWANOperations() throws Exception {
@@ -282,60 +278,85 @@ public class SerialGatewaySenderDistributedDeadlockDUnitTest extends WANTestBase
// messaging between the WAN gateways and members
// exercise region and gateway operations
- vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 100));
- vm5.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 100));
- Wait.pause(2000); // wait for events to propagate
- vm4.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 100));
- vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 100));
- vm5.invoke(() -> WANTestBase.doDestroys(getTestMethodName() + "_RR", 100));
- Wait.pause(2000);// wait for events to propagate
- vm5.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 0));
- vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 0));
- vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 100));
- vm5.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 100));
- Wait.pause(2000); // wait for events to propagate
- vm4.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 100));
- vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 100));
- vm4.invoke(() -> SerialGatewaySenderDistributedDeadlockDUnitTest
- .doInvalidates(getTestMethodName() + "_RR", 100, 100));
- vm4.invoke(() -> WANTestBase.doPutAll(getTestMethodName() + "_RR", 100, 10));
- vm5.invoke(() -> WANTestBase.doPutAll(getTestMethodName() + "_RR", 100, 10));
- Wait.pause(2000);// wait for events to propagate
- vm4.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 1000));
- vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 1000));
- vm4.invoke(() -> WANTestBase.doDestroys(getTestMethodName() + "_RR", 1000));
- Wait.pause(2000);// wait for events to propagate
- vm5.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 0));
- vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 0));
- vm4.invoke(() -> WANTestBase.doPutsPDXSerializable(getTestMethodName() + "_RR", 100));
- Wait.pause(2000);
- vm5.invoke(() -> WANTestBase.validateRegionSize_PDX(getTestMethodName() + "_RR", 100));
- vm2.invoke(() -> WANTestBase.validateRegionSize_PDX(getTestMethodName() + "_RR", 100));
+ vm4.invoke("exerciseWANOperations.doPuts",
+ () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 100));
+ vm5.invoke("exerciseWANOperations.doPuts",
+ () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 100));
+ vm4.invoke("exerciseWANOperations.validateRegionSize",
+ () -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 100));
+ vm2.invoke("exerciseWANOperations.validateRegionSize",
+ () -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 100));
+
+ vm5.invoke("exerciseWANOperations.doDestroys",
+ () -> WANTestBase.doDestroys(getTestMethodName() + "_RR", 100));
+ vm5.invoke("exerciseWANOperations.validateRegionSize",
+ () -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 0));
+ vm2.invoke("exerciseWANOperations.validateRegionSize",
+ () -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 0));
+
+ vm4.invoke("exerciseWANOperations.doPuts2",
+ () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 100));
+ vm5.invoke("exerciseWANOperations.doPuts2",
+ () -> WANTestBase.doPuts(getTestMethodName() + "_RR", 100));
+ vm4.invoke("exerciseWANOperations.validateRegionSize",
+ () -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 100));
+ vm2.invoke("exerciseWANOperations.validateRegionSize",
+ () -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 100));
+
+ vm4.invoke("exerciseWANOperations.doInvalidates",
+ () -> SerialGatewaySenderDistributedDeadlockDUnitTest
+ .doInvalidates(getTestMethodName() + "_RR", 100, 100));
+ vm4.invoke("exerciseWANOperations.doPutAll",
+ () -> WANTestBase.doPutAll(getTestMethodName() + "_RR", 100, 10));
+ vm5.invoke("exerciseWANOperations.doPutAll",
+ () -> WANTestBase.doPutAll(getTestMethodName() + "_RR", 100, 10));
+ vm4.invoke("exerciseWANOperations.validateRegionSize",
+ () -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 1000));
+ vm2.invoke("exerciseWANOperations.validateRegionSize",
+ () -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 1000));
+
+ vm4.invoke("exerciseWANOperations.doDestroys",
+ () -> WANTestBase.doDestroys(getTestMethodName() + "_RR", 1000));
+ vm5.invoke("exerciseWANOperations.validateRegionSize",
+ () -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 0));
+ vm2.invoke("exerciseWANOperations.validateRegionSize",
+ () -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 0));
+
+ vm4.invoke("exerciseWANOperations.doPutsPDXSerializable",
+ () -> WANTestBase.doPutsPDXSerializable(getTestMethodName() + "_RR", 100));
+ vm5.invoke("exerciseWANOperations.validateRegionSize_PDX",
+ () -> WANTestBase.validateRegionSize_PDX(getTestMethodName() + "_RR", 100));
+ vm2.invoke("exerciseWANOperations.validateRegionSize_PDX",
+ () -> WANTestBase.validateRegionSize_PDX(getTestMethodName() + "_RR", 100));
}
private void startSerialSenders() throws Exception {
// get one primary sender on vm4 and another primary on vm5
// the startup order matters here so that primaries are
// on different JVM's
- vm4.invoke(() -> WANTestBase.startSender("ln1"));
+ vm4.invoke("start primary sender", () -> WANTestBase.startSender("ln1"));
- vm5.invoke(() -> WANTestBase.startSender("ln2"));
+ vm5.invoke("start primary sender", () -> WANTestBase.startSender("ln2"));
// start secondaries
- vm5.invoke(() -> WANTestBase.startSender("ln1"));
+ vm5.invoke("start secondary sender", () -> WANTestBase.startSender("ln1"));
- vm4.invoke(() -> WANTestBase.startSender("ln2"));
+ vm4.invoke("start secondary sender", () -> WANTestBase.startSender("ln2"));
}
private void createSerialSenders() throws Exception {
- vm4.invoke(() -> WANTestBase.createSender("ln1", 2, false, 100, 10, false, false, null, true));
+ vm4.invoke("create primary sender",
+ () -> WANTestBase.createSender("ln1", 2, false, 100, 10, false, false, null, true));
- vm5.invoke(() -> WANTestBase.createSender("ln1", 2, false, 100, 10, false, false, null, true));
+ vm5.invoke("create secondary sender",
+ () -> WANTestBase.createSender("ln1", 2, false, 100, 10, false, false, null, true));
- vm4.invoke(() -> WANTestBase.createSender("ln2", 2, false, 100, 10, false, false, null, true));
+ vm4.invoke("create primary sender",
+ () -> WANTestBase.createSender("ln2", 2, false, 100, 10, false, false, null, true));
- vm5.invoke(() -> WANTestBase.createSender("ln2", 2, false, 100, 10, false, false, null, true));
+ vm5.invoke("create secondary sender",
+ () -> WANTestBase.createSender("ln2", 2, false, 100, 10, false, false, null, true));
}
public static void doFunctionPuts(String name, int num, Boolean useThreadOwnedSocket)