You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2020/02/10 08:41:16 UTC
[ignite] branch master updated: IGNITE-12499 Improve node recovery
time with parallel partition state execution - Fixes #7232.
This is an automated email from the ASF dual-hosted git repository.
agoncharuk 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 0e5ba25 IGNITE-12499 Improve node recovery time with parallel partition state execution - Fixes #7232.
0e5ba25 is described below
commit 0e5ba250e1ff19ca8c5556f4ae81838ad391db91
Author: ibessonov <be...@gmail.com>
AuthorDate: Mon Feb 10 11:39:37 2020 +0300
IGNITE-12499 Improve node recovery time with parallel partition state execution - Fixes #7232.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
---
.../processors/cache/GridCacheProcessor.java | 53 ++++++++++++++++++++--
1 file changed, 48 insertions(+), 5 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 7247f7f..1cc78d8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -35,6 +35,8 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -44,6 +46,7 @@ import javax.management.MBeanServer;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteCompute;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteInterruptedException;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.CacheExistsException;
import org.apache.ignite.cache.CacheMode;
@@ -144,6 +147,7 @@ import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
import org.apache.ignite.internal.util.F0;
import org.apache.ignite.internal.util.InitializationProtector;
+import org.apache.ignite.internal.util.StripedExecutor;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -5512,7 +5516,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
/** {@inheritDoc} */
@Override public void afterLogicalUpdatesApplied(
IgniteCacheDatabaseSharedManager mgr,
- GridCacheDatabaseSharedManager.RestoreLogicalState restoreState) throws IgniteCheckedException {
+ GridCacheDatabaseSharedManager.RestoreLogicalState restoreState
+ ) throws IgniteCheckedException {
restorePartitionStates(cacheGroups(), restoreState.partitionRecoveryStates());
}
@@ -5530,15 +5535,53 @@ public class GridCacheProcessor extends GridProcessorAdapter {
if (log.isInfoEnabled())
log.info("Restoring partition state for local groups.");
- long totalProcessed = 0;
+ AtomicLong totalProcessed = new AtomicLong();
+
+ AtomicReference<IgniteCheckedException> restoreStateError = new AtomicReference<>();
+
+ StripedExecutor stripedExec = ctx.getStripedExecutorService();
+
+ int roundRobin = 0;
+
+ for (CacheGroupContext grp : forGroups) {
+ stripedExec.execute(roundRobin % stripedExec.stripesCount(), () -> {
+ try {
+ long processed = grp.offheap().restorePartitionStates(partitionStates);
+
+ totalProcessed.addAndGet(processed);
+ }
+ catch (IgniteCheckedException | RuntimeException | Error e) {
+ U.error(log, "Failed to restore partition state for " +
+ "groupName=" + grp.name() + " groupId=" + grp.groupId(), e);
+
+ restoreStateError.compareAndSet(
+ null,
+ e instanceof IgniteCheckedException
+ ? ((IgniteCheckedException)e)
+ : new IgniteCheckedException(e)
+ );
+ }
+ });
+
+ roundRobin++;
+ }
+
+ try {
+ // Await completion restore state tasks in all stripes.
+ stripedExec.awaitComplete();
+ }
+ catch (InterruptedException e) {
+ throw new IgniteInterruptedException(e);
+ }
- for (CacheGroupContext grp : forGroups)
- totalProcessed += grp.offheap().restorePartitionStates(partitionStates);
+ // Checking error after all task applied.
+ if (restoreStateError.get() != null)
+ throw restoreStateError.get();
if (log.isInfoEnabled())
log.info("Finished restoring partition state for local groups [" +
"groupsProcessed=" + forGroups.size() +
- ", partitionsProcessed=" + totalProcessed +
+ ", partitionsProcessed=" + totalProcessed.get() +
", time=" + (U.currentTimeMillis() - startRestorePart) + "ms]");
}
}