IGNITE-13190

modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java

@@ -81,6 +81,7 @@
 import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition.DFLT_ATOMIC_CACHE_DELETE_HISTORY_SIZE;
 import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition.DFLT_CACHE_REMOVE_ENTRIES_TTL;
 import static org.apache.ignite.internal.processors.cache.mvcc.MvccCachingManager.DFLT_MVCC_TX_SIZE_CACHING_THRESHOLD;
+import static org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager.DFLT_DEFRAGMENTATION_REGION_SIZE_PERCENTAGE;
 import static org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager.DFLT_PDS_WAL_REBALANCE_THRESHOLD;
 import static org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistory.DFLT_PDS_MAX_CHECKPOINT_MEMORY_HISTORY_SIZE;
 import static org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointWorkflow.DFLT_CHECKPOINT_PARALLEL_SORT_THRESHOLD;
@@ -1948,6 +1949,19 @@ public final class IgniteSystemProperties {
         type = Boolean.class)
     public static final String IGNITE_TEST_ENV = "IGNITE_TEST_ENV";
 
+    /**
+     * Defragmentation region size percentage of configured region size.
+     * This percentage will be calculated from largest configured region size and then proportionally subtracted
+     * from all configured regions.
+     */
+    @SystemProperty(value = "Defragmentation region size percentage of configured region size. " +
+        "This percentage will be calculated from largest configured region size and then proportionally subtracted " +
+        "from all configured regions.",
+        type = Integer.class,
+        defaults = "" + DFLT_DEFRAGMENTATION_REGION_SIZE_PERCENTAGE)
+    public static final String IGNITE_DEFRAGMENTATION_REGION_SIZE_PERCENTAGE =
+        "IGNITE_DEFRAGMENTATION_REGION_SIZE_PERCENTAGE";
+
     /**
      * Enforces singleton.
      */

modules/core/src/main/java/org/apache/ignite/internal/maintenance/MaintenanceProcessor.java

@@ -190,7 +190,7 @@ else if (isMaintenanceMode()) {
      */
     private void proceedWithMaintenance() {
         for (Map.Entry<String, MaintenanceWorkflowCallback> cbE : workflowCallbacks.entrySet()) {
-            MaintenanceAction mntcAct = cbE.getValue().automaticAction();
+            MaintenanceAction<?> mntcAct = cbE.getValue().automaticAction();
 
             if (mntcAct != null) {
                 try {

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

@@ -1596,7 +1596,7 @@ void decrementSize(int cacheId) {
                 return grp.mvccEnabled() ? dataTree.isEmpty() : storageSize.get() == 0;
             }
             catch (IgniteCheckedException e) {
-                U.error(log, "Failed to perform operation.", e);
+                U.error(grp.shared().logger(IgniteCacheOffheapManagerImpl.class), "Failed to perform operation.", e);
 
                 return false;
             }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java

@@ -117,7 +117,11 @@
 import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointProgress;
 import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointStatus;
 import org.apache.ignite.internal.processors.cache.persistence.checkpoint.Checkpointer;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.LightweightCheckpointManager;
 import org.apache.ignite.internal.processors.cache.persistence.checkpoint.ReservationReason;
+import org.apache.ignite.internal.processors.cache.persistence.defragmentation.CachePartitionDefragmentationManager;
+import org.apache.ignite.internal.processors.cache.persistence.defragmentation.DefragmentationPageReadWriteManager;
+import org.apache.ignite.internal.processors.cache.persistence.defragmentation.maintenance.DefragmentationWorkflowCallback;
 import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
 import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
 import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStore;
@@ -145,6 +149,7 @@
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.internal.util.typedef.internal.CU;
 import org.apache.ignite.internal.util.typedef.internal.SB;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteBiPredicate;
@@ -162,6 +167,7 @@
 
 import static java.util.Objects.nonNull;
 import static java.util.function.Function.identity;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_DEFRAGMENTATION_REGION_SIZE_PERCENTAGE;
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD;
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_PREFER_WAL_REBALANCE;
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_RECOVERY_SEMAPHORE_PERMITS;
@@ -178,7 +184,10 @@
 import static org.apache.ignite.internal.processors.cache.persistence.CheckpointState.FINISHED;
 import static org.apache.ignite.internal.processors.cache.persistence.CheckpointState.LOCK_RELEASED;
 import static org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointReadWriteLock.CHECKPOINT_LOCK_HOLD_COUNT;
+import static org.apache.ignite.internal.processors.cache.persistence.defragmentation.CachePartitionDefragmentationManager.DEFRAGMENTATION_MNTC_TASK_NAME;
+import static org.apache.ignite.internal.processors.cache.persistence.defragmentation.maintenance.DefragmentationParameters.fromStore;
 import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CORRUPTED_DATA_FILES_MNTC_TASK_NAME;
+import static org.apache.ignite.internal.util.IgniteUtils.GB;
 import static org.apache.ignite.internal.util.IgniteUtils.checkpointBufferSize;
 
 /**
@@ -207,6 +216,12 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
     /** Description of the system view for a {@link MetaStorage}. */
     public static final String METASTORE_VIEW_DESC = "Local metastorage data";
 
+    /** */
+    public static final String DEFRAGMENTATION_PART_REGION_NAME = "defragPartitionsDataRegion";
+
+    /** */
+    public static final String DEFRAGMENTATION_MAPPING_REGION_NAME = "defragMappingDataRegion";
+
     /**
      * Threshold to calculate limit for pages list on-heap caches.
      * <p>
@@ -223,6 +238,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
     /** @see IgniteSystemProperties#IGNITE_PDS_WAL_REBALANCE_THRESHOLD */
     public static final int DFLT_PDS_WAL_REBALANCE_THRESHOLD = 500;
 
+    /** @see IgniteSystemProperties#IGNITE_DEFRAGMENTATION_REGION_SIZE_PERCENTAGE */
+    public static final int DFLT_DEFRAGMENTATION_REGION_SIZE_PERCENTAGE = 60;
+
     /** */
     private final int walRebalanceThreshold =
         getInteger(IGNITE_PDS_WAL_REBALANCE_THRESHOLD, DFLT_PDS_WAL_REBALANCE_THRESHOLD);
@@ -234,6 +252,10 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
     private final String throttlingPolicyOverride = IgniteSystemProperties.getString(
         IgniteSystemProperties.IGNITE_OVERRIDE_WRITE_THROTTLING_ENABLED);
 
+    /** Defragmentation regions size percentage of configured ones. */
+    private final int defragmentationRegionSizePercentageOfConfiguredSize =
+        getInteger(IGNITE_DEFRAGMENTATION_REGION_SIZE_PERCENTAGE, DFLT_DEFRAGMENTATION_REGION_SIZE_PERCENTAGE);
+
     /** */
     private static final String MBEAN_NAME = "DataStorageMetrics";
 
@@ -318,6 +340,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
     /** Lock for releasing history for preloading. */
     private ReentrantLock releaseHistForPreloadingLock = new ReentrantLock();
 
+    /** */
+    private CachePartitionDefragmentationManager defrgMgr;
+
     /** Data regions which should be checkpointed. */
     protected final Set<DataRegion> checkpointedDataRegions = new GridConcurrentHashSet<>();
 
@@ -447,6 +472,32 @@ private DataRegionConfiguration createMetastoreDataRegionConfig(DataStorageConfi
         return cfg;
     }
 
+    /** */
+    private DataRegionConfiguration createDefragmentationDataRegionConfig(long regionSize) {
+        DataRegionConfiguration cfg = new DataRegionConfiguration();
+
+        cfg.setName(DEFRAGMENTATION_PART_REGION_NAME);
+        cfg.setInitialSize(regionSize);
+        cfg.setMaxSize(regionSize);
+        cfg.setPersistenceEnabled(true);
+        cfg.setLazyMemoryAllocation(false);
+
+        return cfg;
+    }
+
+    /** */
+    private DataRegionConfiguration createDefragmentationMappingRegionConfig(long regionSize) {
+        DataRegionConfiguration cfg = new DataRegionConfiguration();
+
+        cfg.setName(DEFRAGMENTATION_MAPPING_REGION_NAME);
+        cfg.setInitialSize(regionSize);
+        cfg.setMaxSize(regionSize);
+        cfg.setPersistenceEnabled(true);
+        cfg.setLazyMemoryAllocation(false);
+
+        return cfg;
+    }
+
     /** {@inheritDoc} */
     @Override protected void start0() throws IgniteCheckedException {
         super.start0();
@@ -497,6 +548,99 @@ private DataRegionConfiguration createMetastoreDataRegionConfig(DataStorageConfi
         }
     }
 
+    /** {@inheritDoc} */
+    @Override protected void initDataRegions(DataStorageConfiguration memCfg) throws IgniteCheckedException {
+        if (isDefragmentationScheduled() && !dataRegionsInitialized) {
+            //Region size configuration will be changed for defragmentation needs.
+            memCfg = configureDataRegionForDefragmentation(memCfg);
+        }
+
+        super.initDataRegions(memCfg);
+    }
+
+    /**
+     * Configure data regions:
+     * <p> Size of configured cache data regions will be decreased in order of freeing space for</p>
+     * <p>defragmentation needs. * New defragmentation regions will be created which size would be based on freed space
+     * from previous step.</p>
+     *
+     * @param memCfg Data storage configuration with data region configurations.
+     * @return New data storage configuration which contains data regions with changed size.
+     * @throws IgniteCheckedException If fail.
+     */
+    private DataStorageConfiguration configureDataRegionForDefragmentation(
+        DataStorageConfiguration memCfg
+    ) throws IgniteCheckedException {
+        List<DataRegionConfiguration> regionConfs = new ArrayList<>();
+
+        DataStorageConfiguration dataConf = memCfg;//not do the changes in-place it's better to make the copy of memCfg.
+
+        regionConfs.add(dataConf.getDefaultDataRegionConfiguration());
+
+        if (dataConf.getDataRegionConfigurations() != null)
+            regionConfs.addAll(Arrays.asList(dataConf.getDataRegionConfigurations()));
+
+        long totalDefrRegionSize = 0;
+        long totalRegionsSize = 0;
+
+        for (DataRegionConfiguration regionCfg : regionConfs) {
+            totalDefrRegionSize = Math.max(
+                totalDefrRegionSize,
+                (long)(regionCfg.getMaxSize() * 0.01 * defragmentationRegionSizePercentageOfConfiguredSize)
+            );
+
+            totalRegionsSize += regionCfg.getMaxSize();
+        }
+
+        double shrinkPercentage = 1d * (totalRegionsSize - totalDefrRegionSize) / totalRegionsSize;
+
+        for (DataRegionConfiguration region : regionConfs) {
+            long newSize = (long)(region.getMaxSize() * shrinkPercentage);
+            long newInitSize = Math.min(region.getInitialSize(), newSize);
+
+            log.info("Region size was reassigned by defragmentation reason: " +
+                "region = '" + region.getName() + "', " +
+                "oldInitialSize = '" + region.getInitialSize() + "', " +
+                "newInitialSize = '" + newInitSize + "', " +
+                "oldMaxSize = '" + region.getMaxSize() + "', " +
+                "newMaxSize = '" + newSize
+            );
+
+            region.setMaxSize(newSize);
+            region.setInitialSize(newInitSize);
+            region.setCheckpointPageBufferSize(0);
+        }
+
+        long mappingRegionSize = Math.min(GB, (long)(totalDefrRegionSize * 0.1));
+
+        checkpointedDataRegions.remove(
+            addDataRegion(
+                memCfg,
+                createDefragmentationDataRegionConfig(totalDefrRegionSize - mappingRegionSize),
+                true,
+                new DefragmentationPageReadWriteManager(cctx.kernalContext(), "defrgPartitionsStore")
+            )
+        );
+
+        checkpointedDataRegions.remove(
+            addDataRegion(
+                memCfg,
+                createDefragmentationMappingRegionConfig(mappingRegionSize),
+                true,
+                new DefragmentationPageReadWriteManager(cctx.kernalContext(), "defrgLinkMappingStore")
+            )
+        );
+
+        return dataConf;
+    }
+
+    /**
+     * @return {@code true} if maintenance mode is on and defragmentation task exists.
+     */
+    private boolean isDefragmentationScheduled() {
+        return cctx.kernalContext().maintenanceRegistry().activeMaintenanceTask(DEFRAGMENTATION_MNTC_TASK_NAME) != null;
+    }
+
     /** */
     public Collection<DataRegion> checkpointedDataRegions() {
         return checkpointedDataRegions;
@@ -603,10 +747,51 @@ private void releaseFileLock() {
         fileLockHolder.close();
     }
 
+    /** */
+    private void prepareCacheDefragmentation(List<Integer> cacheGroupIds) throws IgniteCheckedException {
+        GridKernalContext kernalCtx = cctx.kernalContext();
+        DataStorageConfiguration dsCfg = kernalCtx.config().getDataStorageConfiguration();
+
+        assert CU.isPersistenceEnabled(dsCfg);
+
+        List<DataRegion> regions = Arrays.asList(
+            dataRegion(DEFRAGMENTATION_MAPPING_REGION_NAME),
+            dataRegion(DEFRAGMENTATION_PART_REGION_NAME)
+        );
+
+        LightweightCheckpointManager lightCheckpointMgr = new LightweightCheckpointManager(
+            kernalCtx::log,
+            cctx.igniteInstanceName(),
+            "db-checkpoint-thread-defrag",
+            kernalCtx.workersRegistry(),
+            persistenceCfg,
+            () -> regions,
+            this::getPageMemoryForCacheGroup,
+            resolveThrottlingPolicy(),
+            snapshotMgr,
+            persistentStoreMetricsImpl(),
+            kernalCtx.longJvmPauseDetector(),
+            kernalCtx.failure(),
+            kernalCtx.cache()
+        );
+
+        lightCheckpointMgr.start();
+
+        defrgMgr = new CachePartitionDefragmentationManager(
+            cacheGroupIds,
+            cctx,
+            this,
+            (FilePageStoreManager)cctx.pageStore(),
+            checkpointManager,
+            lightCheckpointMgr,
+            persistenceCfg.getPageSize()
+        );
+    }
+
     /** {@inheritDoc} */
     @Override public DataRegion addDataRegion(DataStorageConfiguration dataStorageCfg, DataRegionConfiguration dataRegionCfg,
-        boolean trackable) throws IgniteCheckedException {
-        DataRegion region = super.addDataRegion(dataStorageCfg, dataRegionCfg, trackable);
+        boolean trackable, PageReadWriteManager pmPageMgr) throws IgniteCheckedException {
+        DataRegion region = super.addDataRegion(dataStorageCfg, dataRegionCfg, trackable, pmPageMgr);
 
         checkpointedDataRegions.add(region);
 
@@ -636,6 +821,16 @@ private void readMetastore() throws IgniteCheckedException {
                 registerSystemView();
 
                 notifyMetastorageReadyForRead();
+
+                cctx.kernalContext().maintenanceRegistry()
+                    .registerWorkflowCallbackIfTaskExists(
+                        DEFRAGMENTATION_MNTC_TASK_NAME,
+                        task -> {
+                            prepareCacheDefragmentation(fromStore(task).cacheGroupIds());
+
+                            return new DefragmentationWorkflowCallback(cctx.kernalContext()::log, defrgMgr);
+                        }
+                    );
             }
             finally {
                 metaStorage = null;

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java

@@ -1357,8 +1357,19 @@ private void saveIndexReencryptionStatus(int grpId) throws IgniteCheckedExceptio
     }
 
     /** */
-    public GridSpinBusyLock busyLock() {
-        return busyLock;
+    public GridCacheDataStore createGridCacheDataStore(
+        CacheGroupContext grpCtx,
+        int partId,
+        boolean exists,
+        IgniteLogger log
+    ) {
+        return new GridCacheDataStore(
+            grpCtx,
+            partId,
+            true,
+            busyLock,
+            log
+        );
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java

@@ -147,7 +147,7 @@ public class IgniteCacheDatabaseSharedManager extends GridCacheSharedManagerAdap
     protected final Map<String, DataRegionMetrics> memMetricsMap = new ConcurrentHashMap<>();
 
     /** */
-    private volatile boolean dataRegionsInitialized;
+    protected volatile boolean dataRegionsInitialized;
 
     /** */
     private volatile boolean dataRegionsStarted;

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/LightweightCheckpointManager.java

@@ -34,7 +34,6 @@
 import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
 import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
 import org.apache.ignite.internal.processors.cache.persistence.DataStorageMetricsImpl;
-import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImpl;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteCacheSnapshotManager;
@@ -106,8 +105,7 @@ public LightweightCheckpointManager(
         DataStorageMetricsImpl persStoreMetrics,
         LongJVMPauseDetector longJvmPauseDetector,
         FailureProcessor failureProcessor,
-        GridCacheProcessor cacheProcessor,
-        FilePageStoreManager pageStoreManager
+        GridCacheProcessor cacheProcessor
     ) throws IgniteCheckedException {
         CheckpointReadWriteLock lock = new CheckpointReadWriteLock(logger);
 
@@ -139,7 +137,7 @@ public LightweightCheckpointManager(
             logger,
             snapshotMgr,
             (pageMemEx, fullPage, buf, tag) ->
-                pageStoreManager.write(fullPage.groupId(), fullPage.pageId(), buf, tag, true),
+                pageMemEx.pageManager().write(fullPage.groupId(), fullPage.pageId(), buf, tag, true),
             persStoreMetrics,
             throttlingPolicy,
             threadBuf,