This is an automated email from the ASF dual-hosted git repository.
sergey-chugunov-1985 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 4cac38bd7e8 IGNITE-28996 Idle Verify utility shouldn't use Java's
ForkJoinPool (#13494)
4cac38bd7e8 is described below
commit 4cac38bd7e817b6ba2d15c584892a3bb85e00f18
Author: Vladimir Steshin <[email protected]>
AuthorDate: Fri Aug 21 11:32:15 2026 +0300
IGNITE-28996 Idle Verify utility shouldn't use Java's ForkJoinPool (#13494)
---
docs/_docs/tools/control-script.adoc | 16 ++++++++
.../apache/ignite/util/GridCommandHandlerTest.java | 10 +++--
.../cache/VerifyBackupPartitionsTask.java | 28 ++++++++------
.../consistency/ConsistencyRepairTask.java | 10 +++--
.../processors/cache/verify/IdleVerifyUtility.java | 4 +-
.../internal/processors/pool/PoolProcessor.java | 43 ++++++++++++++++++++++
6 files changed, 90 insertions(+), 21 deletions(-)
diff --git a/docs/_docs/tools/control-script.adoc
b/docs/_docs/tools/control-script.adoc
index fd96397f261..edc87dfa66d 100644
--- a/docs/_docs/tools/control-script.adoc
+++ b/docs/_docs/tools/control-script.adoc
@@ -755,6 +755,17 @@ To cancel a running `idle_verify` operation, use the
`--cancel` argument:
control.sh --cache idle_verify --cancel
----
+[NOTE]
+====
+[discrete]
+`idle_verify` command uses its own utility thread pool of default size `Total
CPUs - 2` with the minimum of `4` threads.
+The pool size might be adjusted at node startup with the system property
+
+`-DIGNITE_IDLE_VERIFY_POOL_SIZE=...`.
+[discrete]
+These pool threads expire after some time being unused.
+====
+
=== Repairing cache consistency
[WARNING]
====
@@ -823,6 +834,11 @@ control.bat --enable-experimental --consistency status
----
--
+[NOTE]
+====
+Consistency `repair` uses the same thread pool as `idle_verify`.
+====
+
=== Partition update counters finalization
The command allows fo finalize partition update counters after the manual
repair.
diff --git
a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
index 872fb04cce0..38bac2352b1 100644
---
a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
+++
b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
@@ -251,6 +251,8 @@ public class GridCommandHandlerTest extends
GridCommandHandlerClusterPerMethodAb
super.afterTest();
listeningLog = null;
+
+ VerifyBackupPartitionsTask.EXECUTOR_SERVICE = null;
}
/** {@inheritDoc} */
@@ -435,7 +437,7 @@ public class GridCommandHandlerTest extends
GridCommandHandlerClusterPerMethodAb
}, false);
}
- /** */
+ /** TODO: revise in https://issues.apache.org/jira/browse/IGNITE-29002 */
@Test
public void testIdleVerifyCancelBeforeCalcPartitionHashStarted() throws
Exception {
doTestCancelIdleVerify((beforeCancelLatch, afterCancelLatch) -> {
@@ -456,11 +458,11 @@ public class GridCommandHandlerTest extends
GridCommandHandlerClusterPerMethodAb
}
};
- VerifyBackupPartitionsTask.poolSupplier = () -> pool;
+ VerifyBackupPartitionsTask.EXECUTOR_SERVICE = pool;
}, false);
}
- /** */
+ /** TODO: revise in https://issues.apache.org/jira/browse/IGNITE-29002 */
@Test
public void testIdleVerifyCancelWhileCalcPartitionHashRunning() throws
Exception {
for (boolean checkCrc : new boolean[] {false, true}) {
@@ -508,7 +510,7 @@ public class GridCommandHandlerTest extends
GridCommandHandlerClusterPerMethodAb
}
};
- VerifyBackupPartitionsTask.poolSupplier = () -> pool;
+ VerifyBackupPartitionsTask.EXECUTOR_SERVICE = pool;
}, checkCrc);
assertTrue("All tasks must be cancelled",
interruptedOnCancel.get());
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java
b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java
index 54c5f1c41a0..895b5693a41 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java
@@ -31,7 +31,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BooleanSupplier;
-import java.util.function.Supplier;
import java.util.regex.Pattern;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
@@ -57,7 +56,6 @@ import
org.apache.ignite.internal.processors.cache.persistence.file.FilePageStor
import org.apache.ignite.internal.processors.cache.verify.GridNotIdleException;
import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
import org.apache.ignite.internal.processors.task.GridInternal;
-import org.apache.ignite.internal.thread.pool.IgniteForkJoinPool;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteInClosure;
@@ -65,6 +63,7 @@ import org.apache.ignite.resources.IgniteInstanceResource;
import org.apache.ignite.resources.LoggerResource;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.TestOnly;
import static java.util.Collections.emptyMap;
import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_DATA;
@@ -84,6 +83,9 @@ import static
org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtili
*/
@GridInternal
public class VerifyBackupPartitionsTask extends
ComputeTaskAdapter<CacheIdleVerifyCommandArg, IdleVerifyResult> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
/** Error thrown when idle_verify is called on an inactive cluster with
persistence. */
public static final String IDLE_VERIFY_ON_INACTIVE_CLUSTER_ERROR_MESSAGE =
"Cannot perform the operation because " +
"the cluster is inactive.";
@@ -94,16 +96,17 @@ public class VerifyBackupPartitionsTask extends
ComputeTaskAdapter<CacheIdleVeri
/** Checkpoint reason. */
public static final String CP_REASON = "VerifyBackupPartitions";
- /** Shared for tests. */
- public static Supplier<ExecutorService> poolSupplier =
IgniteForkJoinPool::commonPool;
+ /**
+ * Only for tests. Overrides the default pool.
+ * TODO: remove in https://issues.apache.org/jira/browse/IGNITE-29002
+ */
+ @TestOnly
+ public static volatile ExecutorService EXECUTOR_SERVICE;
/** Injected logger. */
@LoggerResource
private IgniteLogger log;
- /** */
- private static final long serialVersionUID = 0L;
-
/** {@inheritDoc} */
@NotNull @Override public Map<? extends ComputeJob, ClusterNode> map(
List<ClusterNode> subgrid,
@@ -335,9 +338,7 @@ public class VerifyBackupPartitionsTask extends
ComputeTaskAdapter<CacheIdleVeri
}
/** */
- private List<Future<Map<PartitionKey, PartitionHashRecord>>>
calcPartitionHashAsync(
- Set<Integer> grpIds
- ) {
+ private List<Future<Map<PartitionKey, PartitionHashRecord>>>
calcPartitionHashAsync(Set<Integer> grpIds) {
List<Future<Map<PartitionKey, PartitionHashRecord>>>
partHashCalcFutures = new ArrayList<>();
for (Integer grpId : grpIds) {
@@ -346,10 +347,13 @@ public class VerifyBackupPartitionsTask extends
ComputeTaskAdapter<CacheIdleVeri
if (grpCtx == null)
continue;
- ExecutorService pool = poolSupplier.get();
+ ExecutorService execSrvs = EXECUTOR_SERVICE;
+
+ if (execSrvs == null)
+ execSrvs =
ignite.context().pools().getIdleVerifyExecutorService();
for (GridDhtLocalPartition part :
grpCtx.topology().currentLocalPartitions())
- partHashCalcFutures.add(calculatePartitionHashAsync(pool,
grpCtx, part, this::isCancelled));
+
partHashCalcFutures.add(calculatePartitionHashAsync(execSrvs, grpCtx, part,
this::isCancelled));
}
return partHashCalcFutures;
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/management/consistency/ConsistencyRepairTask.java
b/modules/core/src/main/java/org/apache/ignite/internal/management/consistency/ConsistencyRepairTask.java
index 0843658042c..1702d1097bd 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/management/consistency/ConsistencyRepairTask.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/management/consistency/ConsistencyRepairTask.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import javax.cache.CacheException;
@@ -40,7 +41,6 @@ import
org.apache.ignite.internal.processors.cache.distributed.dht.topology.Grid
import
org.apache.ignite.internal.processors.cache.distributed.near.consistency.IgniteIrreparableConsistencyViolationException;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.thread.context.Scope;
-import org.apache.ignite.internal.thread.pool.IgniteForkJoinPool;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
import org.apache.ignite.internal.util.lang.GridCursor;
import org.apache.ignite.internal.util.typedef.F;
@@ -102,8 +102,12 @@ public class ConsistencyRepairTask extends
AbstractConsistencyTask<ConsistencyRe
@Override protected String run(ConsistencyRepairCommandArg arg) throws
IgniteException {
AtomicReference<Exception> err = new AtomicReference<>();
+ ExecutorService execSrvc =
ignite.context().pools().getIdleVerifyExecutorService();
+
+ // Consisnency Repair and Idle Verify are usually companion tasks.
Thay share the same thread pool currently
+ // allocated by Idle Verify.
Map<Boolean, List<IgniteBiTuple<Integer, String>>> res =
Arrays.stream(arg.partitions())
- .mapToObj(p -> F.t(p,
IgniteForkJoinPool.commonPool().submit(() -> processPartition(p, arg))))
+ .mapToObj(p -> F.t(p, execSrvc.submit(() ->
processPartition(p, arg))))
.map(t -> {
try {
return F.t(t.get1(), t.get2().get());
@@ -126,7 +130,7 @@ public class ConsistencyRepairTask extends
AbstractConsistencyTask<ConsistencyRe
makeResult(res, false, resStr, CONSISTENCY_VIOLATIONS_FOUND);
- return resStr.length() == 0 ? null : resStr.toString();
+ return resStr.isEmpty() ? null : resStr.toString();
}
/**
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
index 1dc7c5df03e..9e9b2b6a90c 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
@@ -123,7 +123,7 @@ public class IdleVerifyUtility {
for (int pageNo = 0; pageNo < pageStore.pages(); pageId++,
pageNo++) {
if (cancelled != null && cancelled.getAsBoolean())
- throw new IgniteException();
+ throw new IgniteException("Checking of partitions page CRC
sum has been cancelled.");
buf.clear();
@@ -311,7 +311,7 @@ public class IdleVerifyUtility {
while (it.hasNextX()) {
if (cancelled != null && cancelled.getAsBoolean())
- throw new IgniteCheckedException("Caclulate partition hash
cancelled.");
+ throw new IgniteCheckedException("Partition hash calculation
has been cancelled.");
CacheDataRow row = it.nextX();
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
index 27d9642a96d..e6b41bc526c 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
@@ -31,6 +31,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.SystemProperty;
import org.apache.ignite.configuration.ExecutorConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.failure.FailureContext;
@@ -144,6 +145,14 @@ public class PoolProcessor extends GridProcessorAdapter {
*/
public static final long DFLT_PERIODIC_STARVATION_CHECK_FREQ = 30 * 1000L;
+ /** */
+ @SystemProperty(value = "Idle verify utility thread pool size.", type =
Integer.class,
+ defaults = "Total available CPUs minus 2, but at least 4 threads.")
+ public static final String IDLE_VERIFY_POOL_SIZE_PROPERTY =
"IGNITE_IDLE_VERIFY_POOL_SIZE";
+
+ /** */
+ private static final int DFLT_IDLE_VERIFY_POOL_SIZE = Math.max(4,
IgniteConfiguration.AVAILABLE_PROC_CNT - 2);
+
/** Executor service. */
@GridToStringExclude
private IgniteThreadPoolExecutor execSvc;
@@ -164,6 +173,10 @@ public class PoolProcessor extends GridProcessorAdapter {
@GridToStringExclude
private IgniteThreadPoolExecutor mgmtExecSvc;
+ /** Idle verify executor service. */
+ @GridToStringExclude
+ private IgniteThreadPoolExecutor idleVerifyExecSvc;
+
/** P2P executor service. */
@GridToStringExclude
private IgniteThreadPoolExecutor p2pExecSvc;
@@ -377,6 +390,22 @@ public class PoolProcessor extends GridProcessorAdapter {
mgmtExecSvc.allowCoreThreadTimeOut(true);
+ int idleVerifyPoolSz =
IgniteSystemProperties.getInteger(IDLE_VERIFY_POOL_SIZE_PROPERTY,
DFLT_IDLE_VERIFY_POOL_SIZE);
+
+ validateThreadPoolSize(idleVerifyPoolSz, "idle verify");
+
+ idleVerifyExecSvc = new IgniteThreadPoolExecutor(
+ "idleVerify",
+ cfg.getIgniteInstanceName(),
+ idleVerifyPoolSz,
+ idleVerifyPoolSz,
+ DFLT_THREAD_KEEP_ALIVE_TIME,
+ new LinkedBlockingQueue<>(),
+ GridIoPolicy.UNDEFINED,
+ oomeHnd);
+
+ idleVerifyExecSvc.allowCoreThreadTimeOut(true);
+
// Note that since we use 'LinkedBlockingQueue', number of
// maximum threads has no effect.
// Note, that we do not pre-start threads here as class loading pool
may
@@ -622,6 +651,7 @@ public class PoolProcessor extends GridProcessorAdapter {
monitorExecutor("GridSystemExecutor", sysExecSvc);
monitorExecutor("GridClassLoadingExecutor", p2pExecSvc);
monitorExecutor("GridManagementExecutor", mgmtExecSvc);
+ monitorExecutor("GridIdleVerifyExecutor", idleVerifyExecSvc);
monitorExecutor("GridAffinityExecutor", affExecSvc);
monitorExecutor("GridCallbackExecutor", callbackExecSvc);
monitorExecutor("GridQueryExecutor", qryExecSvc);
@@ -898,6 +928,15 @@ public class PoolProcessor extends GridProcessorAdapter {
return mgmtExecSvc;
}
+ /**
+ * Executor service that is in charge to run idle verify routines.
+ *
+ * @return Thread pool implementation to be used idle verify routines.
+ */
+ public IgniteThreadPoolExecutor getIdleVerifyExecutorService() {
+ return idleVerifyExecSvc;
+ }
+
/**
* @return Thread pool implementation to be used for peer class loading
* requests handling.
@@ -1110,6 +1149,10 @@ public class PoolProcessor extends GridProcessorAdapter {
mgmtExecSvc = null;
+ U.shutdownNow(getClass(), idleVerifyExecSvc, log);
+
+ idleVerifyExecSvc = null;
+
U.shutdownNow(getClass(), p2pExecSvc, log);
p2pExecSvc = null;
