Improve node recovery time with parallel partition state execution - Fixes
@@ -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]");
}
}
