You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sk...@apache.org on 2022/04/12 08:28:24 UTC
[ignite] branch master updated: IGNITE-16777 Fixed an issue that could lead to a block of the management pool in case of using clearSync. Fixed #9929
This is an automated email from the ASF dual-hosted git repository.
sk0x50 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 5962b7ab328 IGNITE-16777 Fixed an issue that could lead to a block of the management pool in case of using clearSync. Fixed #9929
5962b7ab328 is described below
commit 5962b7ab3283b7f140bd34b92bf7ad49092fe53b
Author: Denis Chudov <mo...@gmail.com>
AuthorDate: Tue Apr 12 11:28:03 2022 +0300
IGNITE-16777 Fixed an issue that could lead to a block of the management pool in case of using clearSync. Fixed #9929
Signed-off-by: Slava Koptilin <sl...@gmail.com>
---
.../ignite/internal/IgniteInternalFuture.java | 27 ++++
.../processors/cache/GridCacheAdapter.java | 8 +-
.../platform/compute/PlatformCompute.java | 11 ++
.../internal/util/future/GridFinishedFuture.java | 47 +++++++
.../internal/util/future/GridFutureAdapter.java | 63 +++++++++
.../cache/CacheClearAsyncDeadlockTest.java | 75 +++++++++++
.../util/future/GridFutureAdapterSelfTest.java | 148 ++++++++++++++++++++-
.../ignite/testsuites/IgniteCacheTestSuite13.java | 3 +
8 files changed, 369 insertions(+), 13 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteInternalFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteInternalFuture.java
index f7385525acb..3fe750756d2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteInternalFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteInternalFuture.java
@@ -132,6 +132,33 @@ public interface IgniteInternalFuture<R> {
@Async.Schedule
public <T> IgniteInternalFuture<T> chain(IgniteClosure<? super IgniteInternalFuture<R>, T> doneCb, Executor exec);
+ /**
+ * Make a chained future that is completed when {@code doneCb} is executed. Callback is called with this future
+ * as the argument, when this future completes. It is guaranteed that done callback will be called only ONCE.
+ *
+ * @param doneCb Done callback.
+ * @param <T> Type parameter.
+ * @return Chained future.
+ */
+ @Async.Schedule
+ public <T> IgniteInternalFuture<T> chainCompose(
+ IgniteClosure<? super IgniteInternalFuture<R>, IgniteInternalFuture<T>> doneCb
+ );
+
+ /**
+ * Make a chained future that is completed when {@code doneCb} is executed. Callback is called with this future
+ * as the argument, when this future completes. It is guaranteed that done callback will be called only ONCE.
+ *
+ * @param doneCb Done callback.
+ * @param exec Executor to run callback.
+ * @param <T> Type parameter.
+ * @return Chained future.
+ */
+ @Async.Schedule
+ public <T> IgniteInternalFuture<T> chainCompose(
+ IgniteClosure<? super IgniteInternalFuture<R>, IgniteInternalFuture<T>> doneCb, Executor exec
+ );
+
/**
* @return Error value if future has already been completed with error.
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index fe044671a4b..92f2b9628a3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -1228,13 +1228,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
if (isLocal())
return clearLocallyAsync(keys);
else
- return executeClearTask(keys, false).chain(new CX1<IgniteInternalFuture<?>, Object>() {
- @Override public Object applyx(IgniteInternalFuture<?> fut) throws IgniteCheckedException {
- executeClearTask(keys, true).get();
-
- return null;
- }
- });
+ return executeClearTask(keys, false).chainCompose(fut -> executeClearTask(keys, true));
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
index 982f85f57b5..298133233d5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
@@ -44,6 +44,7 @@ import org.apache.ignite.internal.util.future.IgniteFutureImpl;
import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
+import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBGRID;
@@ -487,6 +488,16 @@ public class PlatformCompute extends PlatformAbstractTarget {
throw new UnsupportedOperationException("Chain operation is not supported.");
}
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture chainCompose(IgniteClosure doneCb) {
+ throw new UnsupportedOperationException("Chain compose operation is not supported.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture chainCompose(IgniteClosure doneCb, @Nullable Executor exec) {
+ throw new UnsupportedOperationException("Chain compose operation is not supported.");
+ }
+
/** {@inheritDoc} */
@Override public Throwable error() {
return fut.error();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFinishedFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFinishedFuture.java
index b99a9df2cdc..5368a77fb49 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFinishedFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFinishedFuture.java
@@ -26,6 +26,7 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteInClosure;
+import org.jetbrains.annotations.Nullable;
/**
* Future that is completed at creation time.
@@ -160,6 +161,52 @@ public class GridFinishedFuture<T> implements IgniteInternalFuture<T> {
return fut;
}
+ /** {@inheritDoc} */
+ @Override public <R> IgniteInternalFuture<R> chainCompose(
+ IgniteClosure<? super IgniteInternalFuture<T>, IgniteInternalFuture<R>> doneCb
+ ) {
+ try {
+ return doneCb.apply(this);
+ }
+ catch (GridClosureException e) {
+ return new GridFinishedFuture<>(e.unwrap());
+ }
+ catch (RuntimeException | Error e) {
+ return new GridFinishedFuture<>(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public <R> IgniteInternalFuture<R> chainCompose(
+ IgniteClosure<? super IgniteInternalFuture<T>, IgniteInternalFuture<R>> doneCb,
+ @Nullable Executor exec
+ ) {
+ final GridFutureAdapter<R> res = new GridFutureAdapter<>();
+
+ exec.execute(() -> {
+ IgniteInternalFuture<R> doneCbFut;
+
+ try {
+ doneCbFut = doneCb.apply(this);
+ } catch (GridClosureException e) {
+ doneCbFut = new GridFinishedFuture<>(e.unwrap());
+ } catch (RuntimeException e) {
+ doneCbFut = new GridFinishedFuture<>(e);
+ }
+
+ doneCbFut.listen(f -> {
+ try {
+ res.onDone(f.get(), null);
+ }
+ catch (Exception e) {
+ res.onDone(e);
+ }
+ });
+ });
+
+ return res;
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(GridFinishedFuture.class, this);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFutureAdapter.java
index e62efe2dfd4..bf7c512972e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFutureAdapter.java
@@ -28,6 +28,7 @@ import org.apache.ignite.internal.IgniteFutureCancelledCheckedException;
import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.util.lang.GridClosureException;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -379,6 +380,68 @@ public class GridFutureAdapter<R> implements IgniteInternalFuture<R> {
return fut;
}
+ /** {@inheritDoc} */
+ @Override public <T> IgniteInternalFuture<T> chainCompose(
+ IgniteClosure<? super IgniteInternalFuture<R>, IgniteInternalFuture<T>> doneCb
+ ) {
+ return chainCompose(doneCb, null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> IgniteInternalFuture<T> chainCompose(
+ IgniteClosure<? super IgniteInternalFuture<R>, IgniteInternalFuture<T>> doneCb,
+ @Nullable Executor exec
+ ) {
+ GridFutureAdapter<T> res = new GridFutureAdapter<>();
+
+ if (ignoreInterrupts)
+ res.ignoreInterrupts();
+
+ listen(fut -> {
+ if (exec == null)
+ applyChainComposeCallback(doneCb, fut, res);
+ else {
+ exec.execute(() -> applyChainComposeCallback(doneCb, fut, res));
+ }
+ });
+
+ return res;
+ }
+
+ /**
+ * Apply done callback given for {@link #chainCompose(IgniteClosure)} and
+ * {@link #chainCompose(IgniteClosure, Executor)} methods.
+ *
+ * @param doneCb Callback.
+ * @param fut Future that should be passed to the callback as the argument.
+ * @param chainFuture Chained future.
+ * @param <T> Type parameter.
+ */
+ private <T> void applyChainComposeCallback(
+ IgniteClosure<? super IgniteInternalFuture<R>, IgniteInternalFuture<T>> doneCb,
+ IgniteInternalFuture<R> fut,
+ GridFutureAdapter<T> chainFuture
+ ) {
+ IgniteInternalFuture<T> doneCbFut;
+
+ try {
+ doneCbFut = doneCb.apply(fut);
+ } catch (GridClosureException e) {
+ doneCbFut = new GridFinishedFuture<>(e.unwrap());
+ } catch (RuntimeException e) {
+ doneCbFut = new GridFinishedFuture<>(e);
+ }
+
+ doneCbFut.listen(f -> {
+ try {
+ chainFuture.onDone(f.get(), null);
+ }
+ catch (Exception e) {
+ chainFuture.onDone(e);
+ }
+ });
+ }
+
/**
* @return Logger instance.
*/
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheClearAsyncDeadlockTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheClearAsyncDeadlockTest.java
new file mode 100644
index 00000000000..ce1554c3ea5
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheClearAsyncDeadlockTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.ignite.internal.processors.cache;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+/**
+ * Check that there is no deadlock when clear cache operation is called and management pool is small enough.
+ */
+public class CacheClearAsyncDeadlockTest extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setManagementThreadPoolSize(1)
+ .setCacheConfiguration(
+ new CacheConfiguration<Integer, Integer>(DEFAULT_CACHE_NAME)
+ .setBackups(2)
+ .setNearConfiguration(
+ new NearCacheConfiguration<>()
+ )
+ );
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void test() throws Exception {
+ Ignite ignite = startGridsMultiThreaded(3);
+
+ IgniteCache<Integer, Integer> cache = grid(1).getOrCreateCache(DEFAULT_CACHE_NAME);
+
+ for (int i = 0; i < 1000; i++)
+ cache.put(i, i);
+
+ Ignite client0 = startClientGrid(3);
+
+ IgniteCache<Integer, Integer> clientCache0 = client0.getOrCreateCache(DEFAULT_CACHE_NAME);
+
+ for (int i = 0; i < 1000; i++)
+ clientCache0.get(i);
+
+ Ignite client1 = startClientGrid(4);
+
+ IgniteCache<Integer, Integer> clientCache1 = client1.getOrCreateCache(DEFAULT_CACHE_NAME);
+
+ IgniteFuture<Void> fut = clientCache1.clearAsync();
+
+ fut.get(10_000);
+
+ assertTrue(fut.isDone());
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
index e7e9e5d4ea6..3069a56313f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.util.future;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -33,12 +34,14 @@ import org.apache.ignite.internal.processors.pool.PoolProcessor;
import org.apache.ignite.internal.processors.security.NoOpIgniteSecurityProcessor;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.CX1;
-import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.ignite.thread.IgniteThreadFactory;
import org.junit.Test;
+import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
+
/**
* Tests grid future adapter use cases.
*/
@@ -66,7 +69,7 @@ public class GridFutureAdapterSelfTest extends GridCommonAbstractTest {
final GridFutureAdapter<String> callFut1 = fut;
- GridTestUtils.assertThrows(log, new Callable<Object>() {
+ assertThrows(log, new Callable<Object>() {
@Override public Object call() throws Exception {
return callFut1.get();
}
@@ -78,7 +81,7 @@ public class GridFutureAdapterSelfTest extends GridCommonAbstractTest {
final GridFutureAdapter<String> callFut2 = fut;
- GridTestUtils.assertThrows(log, new Callable<Object>() {
+ assertThrows(log, new Callable<Object>() {
@Override public Object call() throws Exception {
return callFut2.get();
}
@@ -98,7 +101,7 @@ public class GridFutureAdapterSelfTest extends GridCommonAbstractTest {
*/
@Test
public void testOnCancelled() throws Exception {
- GridTestUtils.assertThrows(log, new Callable<Object>() {
+ assertThrows(log, new Callable<Object>() {
@Override public Object call() throws Exception {
GridFutureAdapter<String> fut = new GridFutureAdapter<>();
@@ -108,7 +111,7 @@ public class GridFutureAdapterSelfTest extends GridCommonAbstractTest {
}
}, IgniteFutureCancelledCheckedException.class, null);
- GridTestUtils.assertThrows(log, new Callable<Object>() {
+ assertThrows(log, new Callable<Object>() {
@Override public Object call() throws Exception {
GridFutureAdapter<String> fut = new GridFutureAdapter<>();
@@ -343,6 +346,140 @@ public class GridFutureAdapterSelfTest extends GridCommonAbstractTest {
}
}
+ /**
+ * @throws IgniteCheckedException If failed.
+ */
+ @Test
+ public void testChainCompose() throws IgniteCheckedException {
+ Executor[] executors = new Executor[] { null, Executors.newSingleThreadExecutor() };
+
+ for (Executor exec : executors) {
+ checkCompose(true, false, false, exec);
+ checkCompose(false, false, false, exec);
+ checkCompose(true, true, false, exec);
+ checkCompose(false, true, false, exec);
+
+ checkCompose(true, false, true, exec);
+ checkCompose(false, false, true, exec);
+ checkCompose(true, true, true, exec);
+ checkCompose(false, true, true, exec);
+ }
+
+ checkComposeCancel();
+ }
+
+ /**
+ * @param directOrder Whether to complete the futures in direct order. Direct order is: 1st is completed the future
+ * on which {@link IgniteInternalFuture#chainCompose(IgniteClosure)} is called, 2nd is the future
+ * returned by {@code doneCb} passed to {@link IgniteInternalFuture#chainCompose(IgniteClosure)}.
+ * @param withException Whether the futures are completed with exceptions.
+ * @param cbException Whether the callback throws an exception instead of returning a future.
+ * @param exec Executor.
+ * @throws IgniteCheckedException If failed.
+ */
+ public void checkCompose(
+ boolean directOrder,
+ boolean withException,
+ boolean cbException,
+ Executor exec
+ ) throws IgniteCheckedException {
+ GridFutureAdapter<Object> fut0 = new GridFutureAdapter<>();
+ GridFutureAdapter<Object> cbFut = new GridFutureAdapter<>();
+
+ IgniteInternalFuture<Object> chainedFut = fut0.chainCompose(
+ f -> {
+ if (cbException)
+ throw new RuntimeException("cbException");
+ else
+ return cbFut;
+ },
+ exec
+ );
+
+ assertFalse(chainedFut.isDone());
+
+ Object res0 = withException ? new Exception("test0") : new Object();
+ Object resCb = withException ? new Exception("testCb") : new Object();
+
+ if (directOrder) {
+ futureOnDone(fut0, res0);
+
+ assertTrue(fut0.isDone());
+
+ if (exec == null)
+ assertEquals(cbException, chainedFut.isDone());
+
+ futureOnDone(cbFut, resCb);
+ }
+ else {
+ futureOnDone(cbFut, resCb);
+
+ assertTrue(cbFut.isDone());
+
+ if (exec == null)
+ assertFalse(chainedFut.isDone());
+
+ futureOnDone(fut0, res0);
+ }
+
+ boolean exceptionThrown = false;
+
+ try {
+ chainedFut.get(1000);
+ }
+ catch (Exception e) {
+ exceptionThrown = true;
+
+ if (cbException)
+ assertEquals("cbException", e.getMessage());
+ else if (withException)
+ assertEquals("testCb", e.getMessage());
+ else
+ throw e;
+ }
+
+ assertEquals(withException || cbException, exceptionThrown);
+
+ assertTrue(chainedFut.isDone());
+ }
+
+ /**
+ * Completes the future exceptionally, if {@code res} is the instanse of {@link Throwable}.
+ *
+ * @param fut Future.
+ * @param res Result.
+ */
+ private void futureOnDone(GridFutureAdapter<Object> fut, Object res) {
+ if (res instanceof Throwable) {
+ Throwable t = (Throwable)res;
+
+ fut.onDone(t);
+ }
+ else
+ fut.onDone(res);
+ }
+
+ /**
+ * Check the behavior of compose future when inner future is cancelled.
+ */
+ private void checkComposeCancel() {
+ GridFutureAdapter<Object> fut0 = new GridFutureAdapter<>();
+ GridFutureAdapter<Object> fut1 = new GridFutureAdapter<>();
+
+ IgniteInternalFuture<Object> fut2 = fut0.chainCompose(f -> fut1);
+
+ fut0.onDone(new Object());
+
+ fut1.onCancelled();
+
+ assertTrue(fut1.isDone());
+ assertTrue(fut1.isCancelled());
+
+ assertThrows(log, () -> fut2.get(1000), IgniteFutureCancelledCheckedException.class, null);
+
+ assertTrue(fut2.isDone());
+ }
+
/**
* @throws Exception If failed.
*/
@@ -390,6 +527,5 @@ public class GridFutureAdapterSelfTest extends GridCommonAbstractTest {
catch (IgniteFutureCancelledCheckedException e) {
info("Caught expected exception: " + e);
}
-
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite13.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite13.java
index bae37dc11b1..28a2fc5e189 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite13.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite13.java
@@ -34,6 +34,7 @@ import org.apache.ignite.internal.metric.SystemMetricsTest;
import org.apache.ignite.internal.metric.SystemViewClusterActivationTest;
import org.apache.ignite.internal.metric.SystemViewComputeJobTest;
import org.apache.ignite.internal.metric.SystemViewSelfTest;
+import org.apache.ignite.internal.processors.cache.CacheClearAsyncDeadlockTest;
import org.apache.ignite.internal.processors.cache.GridCacheDataTypesCoverageTest;
import org.apache.ignite.internal.processors.cache.GridCacheLongRunningTransactionDiagnosticsTest;
import org.apache.ignite.internal.processors.cache.GridCacheVersionGenerationWithCacheStorageTest;
@@ -95,6 +96,8 @@ public class IgniteCacheTestSuite13 {
GridTestUtils.addTestIfNeeded(suite, ClusterNameBeforeActivation.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite, CacheClearAsyncDeadlockTest.class, ignoredTests);
+
return suite;
}
}