IGNITE-25736 Speed up warmup running in several threads - Fixes #12153.

Signed-off-by: Aleksey Plekhanov <[email protected]>

Author: skorotkov

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/warmup/LoadAllWarmUpStrategy.java

@@ -22,21 +22,27 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.configuration.DataRegionConfiguration;
 import org.apache.ignite.configuration.LoadAllWarmUpConfiguration;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
 import org.apache.ignite.internal.processors.cache.CacheGroupContext;
 import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
 import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
 import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx;
 import org.apache.ignite.internal.util.tostring.GridToStringExclude;
 import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
 
+import static java.util.stream.Collectors.averagingInt;
 import static java.util.stream.Collectors.toList;
 import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION;
+import static org.apache.ignite.internal.util.IgniteUtils.doInParallel;
 
 /**
  * "Load all" warm-up strategy, which loads pages to persistent data region
@@ -99,12 +105,32 @@ public LoadAllWarmUpStrategy(IgniteLogger log, Supplier<Collection<CacheGroupCon
                 loadDataInfo.keySet().stream().map(CacheGroupContext::cacheOrGroupName).collect(toList()) + ']');
         }
 
-        long loadedPageCnt = 0;
+        AtomicLong loadedPageCnt = new AtomicLong();
 
         for (Map.Entry<CacheGroupContext, List<LoadPartition>> e : loadDataInfo.entrySet()) {
             CacheGroupContext grp = e.getKey();
             List<LoadPartition> parts = e.getValue();
 
+            LoadPartition idxPart = parts.get(0);
+
+            int avgPartPagesCnt = parts.stream()
+                .filter(p -> p.part() != INDEX_PARTITION)
+                .map(LoadPartition::pages)
+                .collect(averagingInt(i -> i)).intValue();
+
+            if (avgPartPagesCnt != 0 && idxPart.pages() > avgPartPagesCnt) {
+                // Split index partition into chunks to balance threads load.  It's needed since
+                // index partition is usually much larger than the data one if indexing is enabled.
+                List<LoadPartition> newParts = new ArrayList<>(idxPart.pages() / avgPartPagesCnt + parts.size());
+
+                for (int i = 0; i < idxPart.pages(); i += avgPartPagesCnt)
+                    newParts.add(new LoadPartition(idxPart.part(), Math.min(idxPart.pages() - i, avgPartPagesCnt), i));
+
+                newParts.addAll(parts.subList(1, parts.size()));
+
+                parts = newParts;
+            }
+
             if (log.isInfoEnabled()) {
                 log.info("Start warm-up cache group, with estimated statistics [name=" + grp.cacheOrGroupName()
                     + ", partCnt=" + parts.size() + ", pageCnt="
@@ -113,31 +139,42 @@ public LoadAllWarmUpStrategy(IgniteLogger log, Supplier<Collection<CacheGroupCon
 
             PageMemoryEx pageMemEx = (PageMemoryEx)region.pageMemory();
 
-            for (LoadPartition part : parts) {
-                long pageId = pageMemEx.partitionMetaPageId(grp.groupId(), part.part());
+            GridKernalContext ctx = grp.shared().kernalContext();
 
-                for (int i = 0; i < part.pages(); i++, pageId++, loadedPageCnt++) {
-                    if (stop) {
-                        if (log.isInfoEnabled()) {
-                            log.info("Stop warm-up cache group with loaded statistics [name="
-                                + grp.cacheOrGroupName() + ", pageCnt=" + loadedPageCnt
-                                + ", remainingPageCnt=" + (availableLoadPageCnt - loadedPageCnt) + ']');
-                        }
+            doInParallel(
+                U.availableThreadCount(ctx, GridIoPolicy.SYSTEM_POOL, 2),
+                ctx.pools().getSystemExecutorService(),
+                parts,
+                part -> {
+                    long pageId = pageMemEx.partitionMetaPageId(grp.groupId(), part.part());
 
-                        return;
-                    }
+                    pageId += part.startPageIdx();
 
-                    long pagePtr = -1;
+                    for (int i = 0; i < part.pages(); i++, pageId++, loadedPageCnt.incrementAndGet()) {
+                        if (stop) {
+                            if (log.isInfoEnabled()) {
+                                log.info("Stop warm-up cache group with loaded statistics [name="
+                                    + grp.cacheOrGroupName() + ", pageCnt=" + loadedPageCnt.get()
+                                    + ", remainingPageCnt=" + (availableLoadPageCnt - loadedPageCnt.get()) + ']');
+                            }
 
-                    try {
-                        pagePtr = pageMemEx.acquirePage(grp.groupId(), pageId);
-                    }
-                    finally {
-                        if (pagePtr != -1)
-                            pageMemEx.releasePage(grp.groupId(), pageId, pagePtr);
+                            return null;
+                        }
+
+                        long pagePtr = -1;
+
+                        try {
+                            pagePtr = pageMemEx.acquirePage(grp.groupId(), pageId);
+                        }
+                        finally {
+                            if (pagePtr != -1)
+                                pageMemEx.releasePage(grp.groupId(), pageId, pagePtr);
+                        }
                     }
+
+                    return null;
                 }
-            }
+            );
         }
     }
 
@@ -194,14 +231,14 @@ protected Map<CacheGroupContext, List<LoadPartition>> loadDataInfo(
             for (int j = -1; j < locParts.size() && availableLoadPageCnt > 0; j++) {
                 int p = j == -1 ? INDEX_PARTITION : locParts.get(j).id();
 
-                long partPageCnt = grp.shared().pageStore().pages(grp.groupId(), p);
+                int partPageCnt = grp.shared().pageStore().pages(grp.groupId(), p);
 
                 if (partPageCnt > 0) {
-                    long pageCnt = (availableLoadPageCnt - partPageCnt) >= 0 ? partPageCnt : availableLoadPageCnt;
+                    int pageCnt = (availableLoadPageCnt - partPageCnt) >= 0 ? partPageCnt : (int)availableLoadPageCnt;
 
                     availableLoadPageCnt -= pageCnt;
 
-                    loadableGrps.computeIfAbsent(grp, grpCtx -> new ArrayList<>()).add(new LoadPartition(p, pageCnt));
+                    loadableGrps.computeIfAbsent(grp, grpCtx -> new ArrayList<>()).add(new LoadPartition(p, pageCnt, 0));
                 }
             }
         }
@@ -217,20 +254,25 @@ static class LoadPartition {
         private final int part;
 
         /** Number of pages to load. */
-        private final long pages;
+        private final int pages;
+
+        /** Index of first page to load. */
+        private final int startPageIdx;
 
         /**
          * Constructor.
          *
          * @param part Partition id.
          * @param pages Number of pages to load.
+         * @param startPageIdx Index of first page to load.
          */
-        public LoadPartition(int part, long pages) {
+        public LoadPartition(int part, int pages, int startPageIdx) {
             assert part >= 0 : "Partition id cannot be negative.";
             assert pages > 0 : "Number of pages to load must be greater than zero.";
 
             this.part = part;
             this.pages = pages;
+            this.startPageIdx = startPageIdx;
         }
 
         /**
@@ -247,10 +289,19 @@ public int part() {
          *
          * @return Number of pages to load.
          */
-        public long pages() {
+        public int pages() {
             return pages;
         }
 
+        /**
+         * Return index of first page to load.
+         *
+         * @return Index of first page to load.
+         */
+        public int startPageIdx() {
+            return startPageIdx;
+        }
+
         /** {@inheritDoc} */
         @Override public String toString() {
             return S.toString(LoadPartition.class, this);

modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/data_storage.py

@@ -19,6 +19,8 @@
 
 from typing import NamedTuple
 
+from ignitetest.utils.bean import Bean
+
 
 class DataRegionConfiguration(NamedTuple):
     """
@@ -33,6 +35,7 @@ class DataRegionConfiguration(NamedTuple):
     metrics_enabled: bool = True
     metrics_rate_time_interval: int = None
     checkpoint_page_buffer_size: int = None
+    warm_up_configuration: Bean = None
 
 
 class DataStorageConfiguration(NamedTuple):
@@ -60,3 +63,4 @@ class DataStorageConfiguration(NamedTuple):
     write_recovery_data_on_checkpoint: bool = None
     checkpoint_recovery_data_compression: str = None
     checkpoint_recovery_data_compression_level: int = None
+    default_warm_up_configuration: Bean = None

modules/ducktests/tests/ignitetest/services/utils/templates/datastorage_macro.j2

@@ -15,6 +15,8 @@
  limitations under the License.
 #}
 
+{% import 'misc_macro.j2' as misc_utils %}
+
 {% macro data_storage(config) %}
     {% if config %}
         <property name="dataStorageConfiguration">
@@ -33,6 +35,10 @@
                             {% endfor %}
                             </list>
                         </property>
+                    {% elif name == 'default_warm_up_configuration' %}
+                        <property name="defaultWarmUpConfiguration">
+                            {{ misc_utils.bean(value) }}
+                        </property>
                     {% else %}
                         <property name="{{ name | snake_to_camel }}" value="{{ value }}"/>
                     {% endif %}
@@ -47,7 +53,13 @@
     <bean class="org.apache.ignite.configuration.DataRegionConfiguration">
         {% for name, value in config._asdict().items() %}
         {% if value != None %}
-        <property name="{{ name | snake_to_camel }}" value="{{ value }}"/>
+            {% if name == 'warm_up_configuration' %}
+                <property name="warmUpConfiguration">
+                    {{ misc_utils.bean(value) }}
+                </property>
+            {% else %}
+                <property name="{{ name | snake_to_camel }}" value="{{ value }}"/>
+            {% endif %}
         {% endif %}
         {% endfor %}
     </bean>

modules/indexing/src/test/java/org/apache/ignite/testsuites/IgnitePdsWithIndexingTestSuite.java

@@ -32,6 +32,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.db.IndexingMultithreadedLoadContinuousRestartTest;
 import org.apache.ignite.internal.processors.cache.persistence.db.LongDestroyDurableBackgroundTaskTest;
 import org.apache.ignite.internal.processors.cache.persistence.db.MultipleParallelCacheDeleteDeadlockTest;
+import org.apache.ignite.internal.processors.cache.warmup.LoadAllWarmUpStrategySelfTest;
 import org.apache.ignite.internal.processors.database.IgniteDbMultiNodeWithIndexingPutGetTest;
 import org.apache.ignite.internal.processors.database.IgniteDbSingleNodeWithIndexingPutGetTest;
 import org.apache.ignite.internal.processors.database.IgniteDbSingleNodeWithIndexingWalRestoreTest;
@@ -75,6 +76,7 @@
     DropIndexTest.class,
     MaintenanceRebuildIndexUtilsSelfTest.class,
     IndexWithSameNameH2Test.class,
+    LoadAllWarmUpStrategySelfTest.class
 })
 public class IgnitePdsWithIndexingTestSuite {
 }