You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by al...@apache.org on 2022/08/18 07:42:33 UTC
[geode] branch develop updated: GEODE-10403: Fix distributed deadlock with stop gw sender (#7830)
This is an automated email from the ASF dual-hosted git repository.
alberto 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 6c257d7c28 GEODE-10403: Fix distributed deadlock with stop gw sender (#7830)
6c257d7c28 is described below
commit 6c257d7c284ffc9328905ae9ca8fd786b9428ede
Author: Alberto Gomez <al...@est.tech>
AuthorDate: Thu Aug 18 09:42:27 2022 +0200
GEODE-10403: Fix distributed deadlock with stop gw sender (#7830)
There is a distributed deadlock that can appear
when stopping the gateway sender if a race condition
happens in which the stop gateway sender command gets blocked
indefinitely trying to get the size of the queue from remote peers
(ParallelGatewaySenderQueue.size() call) and
also one call to store one event in the queue tries to get
the lifecycle lock (acquired by the gateway sender command).
These two calls could get into a deadlock under heavy load and
make the system unresponsive for any traffic request (get, put, ...).
In order to avoid it, in the storage of the event in the gateway
sender queue (AbstractGatewaySender.distribute() call),
instead to trying to get the lifecycle lock without
any timeout, a try with a timeout is added. If the
try returns false it is checked if the gateway sender is running. If
it is not running, the event is dropped and there is no need to get the lock.
Otherwise, the lifecycle lock acquire is retried until it succeeds or
the gateway sender is stopped.
---
.../internal/ClusterOperationExecutors.java | 9 ++-
.../internal/cache/wan/AbstractGatewaySender.java | 52 +++++++++----
.../geode/internal/cache/wan/WANTestBase.java | 7 ++
...llelGatewaySenderOperationsDistributedTest.java | 86 +++++++++++++++++++++-
4 files changed, 133 insertions(+), 21 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterOperationExecutors.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterOperationExecutors.java
index ba25e3b539..7c45bbb9e3 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterOperationExecutors.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterOperationExecutors.java
@@ -25,12 +25,14 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.SystemFailure;
+import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.gms.messages.ViewAckMessage;
import org.apache.geode.internal.logging.CoreLoggingExecutors;
@@ -167,6 +169,8 @@ public class ClusterOperationExecutors implements OperationExecutors {
private SerialQueuedExecutorPool serialQueuedExecutorPool;
+ @MutableForTesting
+ public static final AtomicInteger maxPrThreadsForTest = new AtomicInteger(-1);
ClusterOperationExecutors(DistributionStats stats,
InternalDistributedSystem system) {
@@ -252,10 +256,11 @@ public class ClusterOperationExecutors implements OperationExecutors {
this::doWaitingThread, stats.getWaitingPoolHelper(),
threadMonitor);
- if (MAX_PR_THREADS > 1) {
+ int maxPrThreads = maxPrThreadsForTest.get() > 0 ? maxPrThreadsForTest.get() : MAX_PR_THREADS;
+ if (maxPrThreads > 1) {
partitionedRegionPool =
CoreLoggingExecutors.newThreadPoolWithFeedStatistics(
- MAX_PR_THREADS, INCOMING_QUEUE_LIMIT, stats.getPartitionedRegionQueueHelper(),
+ maxPrThreads, INCOMING_QUEUE_LIMIT, stats.getPartitionedRegionQueueHelper(),
"PartitionedRegion Message Processor",
thread -> stats.incPartitionedRegionThreadStarts(), this::doPartitionRegionThread,
stats.getPartitionedRegionPoolHelper(),
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
index bd1a898da4..09dc843f3a 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
@@ -25,6 +25,7 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.logging.log4j.Logger;
@@ -238,6 +239,9 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
protected boolean enforceThreadsConnectSameReceiver;
+ @MutableForTesting
+ public static final AtomicBoolean doSleepForTestingInDistribute = new AtomicBoolean(false);
+
protected AbstractGatewaySender() {
statisticsClock = disabledClock();
}
@@ -1125,16 +1129,17 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
}
// If this gateway is not running, return
- if (!isRunning()) {
- if (isPrimary()) {
- recordDroppedEvent(clonedEvent);
- }
- if (isDebugEnabled) {
- logger.debug("Returning back without putting into the gateway sender queue:" + event);
- }
+ if (!getIsRunningAndDropEventIfNotRunning(event, isDebugEnabled, clonedEvent)) {
return;
}
+ if (AbstractGatewaySender.doSleepForTestingInDistribute.get()) {
+ try {
+ Thread.sleep(5);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
if (!getLifeCycleLock().readLock().tryLock()) {
synchronized (queuedEventsSync) {
if (!enqueuedAllTempQueueEvents) {
@@ -1151,19 +1156,22 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
}
}
if (enqueuedAllTempQueueEvents) {
- getLifeCycleLock().readLock().lock();
+ try {
+ while (!getLifeCycleLock().readLock().tryLock(10, TimeUnit.MILLISECONDS)) {
+ if (!getIsRunningAndDropEventIfNotRunning(event, isDebugEnabled, clonedEvent)) {
+ return;
+ }
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
}
}
try {
// If this gateway is not running, return
// The sender may have stopped, after we have checked the status in the beginning.
- if (!isRunning()) {
- if (isDebugEnabled) {
- logger.debug("Returning back without putting into the gateway sender queue:" + event);
- }
- if (isPrimary()) {
- recordDroppedEvent(clonedEvent);
- }
+ if (!getIsRunningAndDropEventIfNotRunning(event, isDebugEnabled, clonedEvent)) {
return;
}
@@ -1208,6 +1216,20 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
}
}
+ private boolean getIsRunningAndDropEventIfNotRunning(EntryEventImpl event, boolean isDebugEnabled,
+ EntryEventImpl clonedEvent) {
+ if (isRunning()) {
+ return true;
+ }
+ if (isPrimary()) {
+ recordDroppedEvent(clonedEvent);
+ }
+ if (isDebugEnabled) {
+ logger.debug("Returning back without putting into the gateway sender queue:" + event);
+ }
+ return false;
+ }
+
private void recordDroppedEvent(EntryEventImpl event) {
if (eventProcessor != null) {
eventProcessor.registerEventDroppedInPrimaryQueue(event);
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 2969a87296..aaa8551c07 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
@@ -917,6 +917,13 @@ public class WANTestBase extends DistributedTestCase {
createCache(true, locPort);
}
+ public static void createCacheConserveSocketsInVMs(Boolean conserveSockets, Integer locPort,
+ VM... vms) {
+ for (VM vm : vms) {
+ vm.invoke(() -> createCacheConserveSockets(conserveSockets, locPort));
+ }
+ }
+
public static void createCacheConserveSockets(Boolean conserveSockets, Integer locPort) {
WANTestBase test = new WANTestBase();
Properties props = test.getDistributedSystemProperties();
diff --git a/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDistributedTest.java b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDistributedTest.java
index bc7ea22b21..4e3e4850d3 100644
--- a/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDistributedTest.java
+++ b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDistributedTest.java
@@ -59,6 +59,7 @@ import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.wan.GatewayEventFilter;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
+import org.apache.geode.distributed.internal.ClusterOperationExecutors;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.DistributionMessageObserver;
import org.apache.geode.internal.cache.BucketRegion;
@@ -346,6 +347,66 @@ public class ParallelGatewaySenderOperationsDistributedTest extends WANTestBase
vm2.invoke(() -> validateRegionSizeRemainsSame(getUniqueName() + "_PR", 100));
}
+ /**
+ * Verifies that no distributed deadlock occurs when stopping a gateway sender while receiving
+ * traffic.
+ * The distributed deadlock may occur when the gateway sender tries to get the
+ * size of the gateway sender queue (sending a size message to other members) while holding the
+ * lifeCycleLock lock. This lock is also taken when an event is to be distributed by the gateway
+ * sender.
+ * As this issue has only been observed in the field with a lot of traffic, in order to reproduce
+ * it in a test case, conserve-sockets is set to true (although the deadlock has also
+ * been seen with conserve-sockets=false), the size of the PartitionedRegion thread pool is set
+ * to a small value and an artificial timeout is added at a point in the distribute() call
+ * of the AbstractGatewaySeder class.
+ */
+ @Test
+ public void testNoDistributedDeadlockWithGatewaySenderStop() throws Exception {
+ addIgnoredException("Broken pipe");
+ Integer[] locatorPorts = createLNAndNYLocators();
+ Integer lnPort = locatorPorts[0];
+ Integer nyPort = locatorPorts[1];
+ VM[] senders = {vm4, vm5, vm6, vm7};
+ try {
+ for (VM sender : senders) {
+ sender.invoke(() -> AbstractGatewaySender.doSleepForTestingInDistribute.set(true));
+ sender.invoke(() -> ClusterOperationExecutors.maxPrThreadsForTest.set(2));
+ }
+ vm2.invoke(() -> ClusterOperationExecutors.maxPrThreadsForTest.set(2));
+ vm3.invoke(() -> ClusterOperationExecutors.maxPrThreadsForTest.set(2));
+
+ createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, true, true);
+
+ // make sure all the senders are running before doing any puts
+ waitForSendersRunning();
+
+ // Send a fairly big amount of operations to provoke the deadlock
+ int invocationsPerServer = 4;
+ AsyncInvocation[] invocations = new AsyncInvocation[senders.length * invocationsPerServer];
+ for (int i = 0; i < senders.length; i++) {
+ for (int j = 0; j < invocationsPerServer; j++) {
+ invocations[i + (j * invocationsPerServer)] =
+ senders[i].invokeAsync(() -> doPuts(getUniqueName() + "_PR", 100));
+ }
+ }
+
+ // Wait for some elements to be replicated before stopping the senders
+ for (int i = 0; i < senders.length; i++) {
+ senders[i].invoke(() -> await()
+ .untilAsserted(() -> assertThat(getSenderStats("ln", -1).get(3)).isGreaterThan(1)));
+ }
+
+ stopSendersAsync();
+ for (int i = 0; i < invocations.length; i++) {
+ invocations[i].await();
+ }
+ } finally {
+ for (int i = 0; i < senders.length; i++) {
+ senders[i].invoke(() -> AbstractGatewaySender.doSleepForTestingInDistribute.set(false));
+ }
+ }
+ }
+
/**
* Normal scenario in which a sender is stopped and then started again.
*/
@@ -1271,7 +1332,13 @@ public class ParallelGatewaySenderOperationsDistributedTest extends WANTestBase
private void createSendersReceiversAndPartitionedRegion(Integer lnPort, Integer nyPort,
boolean createAccessors, boolean startSenders) {
- createSendersAndReceivers(lnPort, nyPort);
+ createSendersReceiversAndPartitionedRegion(lnPort, nyPort, createAccessors, startSenders,
+ false);
+ }
+
+ private void createSendersReceiversAndPartitionedRegion(Integer lnPort, Integer nyPort,
+ boolean createAccessors, boolean startSenders, boolean conserveSockets) {
+ createSendersAndReceivers(lnPort, nyPort, conserveSockets);
createPartitionedRegions(createAccessors);
@@ -1280,11 +1347,11 @@ public class ParallelGatewaySenderOperationsDistributedTest extends WANTestBase
}
}
- private void createSendersAndReceivers(Integer lnPort, Integer nyPort) {
- createCacheInVMs(nyPort, vm2, vm3);
+ private void createSendersAndReceivers(Integer lnPort, Integer nyPort, boolean conserveSockets) {
+ createCacheConserveSocketsInVMs(conserveSockets, nyPort, vm2, vm3);
createReceiverInVMs(vm2, vm3);
- createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+ createCacheConserveSocketsInVMs(conserveSockets, lnPort, vm4, vm5, vm6, vm7);
vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true));
vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true));
@@ -1578,6 +1645,17 @@ public class ParallelGatewaySenderOperationsDistributedTest extends WANTestBase
vm7.invoke(() -> stopSender("ln"));
}
+ private void stopSendersAsync() throws InterruptedException {
+ AsyncInvocation inv1 = vm4.invokeAsync(() -> stopSender("ln"));
+ AsyncInvocation inv2 = vm5.invokeAsync(() -> stopSender("ln"));
+ AsyncInvocation inv3 = vm6.invokeAsync(() -> stopSender("ln"));
+ AsyncInvocation inv4 = vm7.invokeAsync(() -> stopSender("ln"));
+ inv1.await();
+ inv2.await();
+ inv3.await();
+ inv4.await();
+ }
+
private void waitForSendersRunning() {
vm4.invoke(() -> waitForSenderRunningState("ln"));
vm5.invoke(() -> waitForSenderRunningState("ln"));