IGNITE-24548 Enable speed-based throttling

modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemory.java

@@ -297,8 +297,7 @@ private void initMetrics() {
      *
      * @param writeThrottle Page write throttling instance.
      */
-    // TODO IGNITE-24548 Make a proper implementation.
-    @TestOnly
+    // TODO IGNITE-24933 Remove this method.
     public void initThrottling(PagesWriteThrottlePolicy writeThrottle) {
         this.writeThrottle = writeThrottle;
     }

modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointManager.java

@@ -242,6 +242,10 @@ public CheckpointProgress scheduleCheckpoint(long delayMillis, String reason) {
         return checkpointer.currentCheckpointProgress();
     }
 
+    public @Nullable CheckpointProgress currentCheckpointProgressForThrottling() {
+        return checkpointer.currentCheckpointProgressForThrottling();
+    }
+
     /**
      * Returns the progress of the last checkpoint, or the current checkpoint if in progress, {@code null} if no checkpoint has occurred.
      */

modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/Checkpointer.java

@@ -152,6 +152,13 @@ public class Checkpointer extends IgniteWorker {
     /** Current checkpoint progress. This field is updated only by checkpoint thread. */
     private volatile @Nullable CheckpointProgressImpl currentCheckpointProgress;
 
+    /**
+     * Checkpoint progress instance with a more limited range of visibility. It is initialized when checkpoint write lick is acquired, and
+     * nullified when checkpoint finishes (unlike {@link #currentCheckpointProgress} that is updated before we started notifying checkpoint
+     * listeners and is never nullified).
+     */
+    private volatile @Nullable CheckpointProgressImpl currentCheckpointProgressForThrottling;
+
     /** Checkpoint progress after releasing write lock. */
     private volatile @Nullable CheckpointProgressImpl afterReleaseWriteLockCheckpointProgress;
 
@@ -418,6 +425,8 @@ void doCheckpoint() throws IgniteInternalCheckedException {
             failureManager.process(new FailureContext(CRITICAL_ERROR, e));
 
             throw e;
+        } finally {
+            currentCheckpointProgressForThrottling = null;
         }
     }
 
@@ -684,6 +693,10 @@ void startCheckpointProgress() {
             scheduledCheckpointProgress = new CheckpointProgressImpl(MILLISECONDS.toNanos(nextCheckpointInterval()));
 
             currentCheckpointProgress = curr;
+
+            curr.futureFor(LOCK_TAKEN).thenRun(() -> {
+                currentCheckpointProgressForThrottling = curr;
+            });
         }
     }
 
@@ -779,6 +792,10 @@ public void shutdownCheckpointer(boolean shutdown) {
         return currentCheckpointProgress;
     }
 
+    public @Nullable CheckpointProgress currentCheckpointProgressForThrottling() {
+        return currentCheckpointProgressForThrottling;
+    }
+
     /**
      * Returns the progress of the last checkpoint, or the current checkpoint if in progress, {@code null} if no checkpoint has occurred.
      */

modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/throttling/PagesWriteSpeedBasedThrottle.java

@@ -53,6 +53,8 @@ public class PagesWriteSpeedBasedThrottle implements PagesWriteThrottlePolicy {
      */
     static final long NO_THROTTLING_MARKER = Long.MIN_VALUE;
 
+    private final long logThresholdNanos;
+
     private final PersistentPageMemory pageMemory;
 
     private final Supplier<CheckpointProgress> cpProgress;
@@ -95,17 +97,20 @@ public class PagesWriteSpeedBasedThrottle implements PagesWriteThrottlePolicy {
     /**
      * Constructor.
      *
+     * @param logThresholdNanos Minimal throttling duration required for printing a warning message to the log.
      * @param pageMemory Page memory.
      * @param cpProgress Database manager.
      * @param stateChecker Checkpoint lock state provider.
      * @param metricSource Metric source.
      */
     public PagesWriteSpeedBasedThrottle(
+            long logThresholdNanos,
             PersistentPageMemory pageMemory,
             Supplier<CheckpointProgress> cpProgress,
             CheckpointLockStateChecker stateChecker,
             PersistentPageMemoryMetricSource metricSource
     ) {
+        this.logThresholdNanos = logThresholdNanos;
         this.pageMemory = pageMemory;
         this.cpProgress = cpProgress;
         cpLockStateChecker = stateChecker;
@@ -187,7 +192,7 @@ private void initMetrics(PersistentPageMemoryMetricSource metricSource) {
             return;
         }
 
-        if (throttleParkTimeNs > LOGGING_THRESHOLD) {
+        if (throttleParkTimeNs > logThresholdNanos) {
             LOG.warn("Parking thread={} for timeout(ms)={}", Thread.currentThread().getName(), throttleParkTimeNs / 1_000_000);
         }
 

modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/throttling/PagesWriteThrottlePolicy.java

@@ -60,9 +60,8 @@
  * </ul>
  */
 public interface PagesWriteThrottlePolicy {
-    // TODO Maybe make it configurable in IGNITE-24548
     /** Min park time which triggers logging. */
-    long LOGGING_THRESHOLD = TimeUnit.SECONDS.toNanos(1);
+    long DEFAULT_LOGGING_THRESHOLD = TimeUnit.SECONDS.toNanos(1);
 
     /** Checkpoint buffer fullfill upper bound. */
     float CP_BUF_FILL_THRESHOLD = 2f / 3;

modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/throttling/SpeedBasedMemoryConsumptionThrottlingStrategy.java

@@ -187,17 +187,19 @@ private long speedBasedParkTime(int cpWrittenPages, long donePages, int cpTotalP
         detectCpPagesWriteStart(cpWrittenPages, dirtyPagesRatio);
 
         if (dirtyPagesRatio >= MAX_DIRTY_PAGES) {
-            return 0; // too late to throttle, will wait on safe to update instead.
+            return 0; // Too late to throttle, will wait on safe to update instead.
         } else {
             return getParkTime(dirtyPagesRatio,
                     donePages,
-                    notEvictedPagesTotal(cpTotalPages),
+                    // TODO IGNITE-24937 Should be a "notEvictedPagesTotal(cpTotalPages)" call.
+                    cpTotalPages,
                     threadIds.size(),
                     instantaneousMarkDirtySpeed,
                     avgCpWriteSpeed);
         }
     }
 
+    // TODO IGNITE-24937 Leads to negative estimations in some cases. Should be fixed.
     private int notEvictedPagesTotal(int cpTotalPages) {
         return Math.max(cpTotalPages - cpEvictedPages(), 0);
     }
@@ -375,7 +377,13 @@ long getCpWriteSpeed() {
      * Returns counter for fsynced checkpoint pages.
      */
     int cpSyncedPages() {
-        AtomicInteger syncedPagesCounter = cpProgress.get().syncedPagesCounter();
+        CheckpointProgress progress = cpProgress.get();
+
+        if (progress == null) {
+            return 0;
+        }
+
+        AtomicInteger syncedPagesCounter = progress.syncedPagesCounter();
 
         // Null-check simplifies testing, we don't have to mock this counter.
         return syncedPagesCounter == null ? 0 : syncedPagesCounter.get();
@@ -385,14 +393,26 @@ int cpSyncedPages() {
      * Return a number of pages in current checkpoint.
      */
     int cpTotalPages() {
-        return cpProgress.get().currentCheckpointPagesCount();
+        CheckpointProgress progress = cpProgress.get();
+
+        if (progress == null) {
+            return 0;
+        }
+
+        return progress.currentCheckpointPagesCount();
     }
 
     /**
      * Returns a number of evicted pages.
      */
     int cpEvictedPages() {
-        AtomicInteger evictedPagesCounter = cpProgress.get().evictedPagesCounter();
+        CheckpointProgress progress = cpProgress.get();
+
+        if (progress == null) {
+            return 0;
+        }
+
+        AtomicInteger evictedPagesCounter = progress.evictedPagesCounter();
 
         // Null-check simplifies testing, we don't have to mock this counter.
         return evictedPagesCounter == null ? 0 : evictedPagesCounter.get();

modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/throttling/TargetRatioPagesWriteThrottle.java

@@ -36,6 +36,8 @@ public class TargetRatioPagesWriteThrottle implements PagesWriteThrottlePolicy {
     /** Logger. */
     private static final IgniteLogger LOG = Loggers.forClass(TargetRatioPagesWriteThrottle.class);
 
+    private final long logThresholdNanos;
+
     /** Page memory. */
     private final PersistentPageMemory pageMemory;
 
@@ -68,17 +70,20 @@ public class TargetRatioPagesWriteThrottle implements PagesWriteThrottlePolicy {
     /**
      * Constructor.
      *
+     * @param logThresholdNanos Minimal throttling duration required for printing a warning message to the log.
      * @param pageMemory Page memory.
      * @param cpProgress Database manager.
      * @param stateChecker checkpoint lock state checker.
      * @param metricSource Metric source.
      */
     public TargetRatioPagesWriteThrottle(
+            long logThresholdNanos,
             PersistentPageMemory pageMemory,
             Supplier<CheckpointProgress> cpProgress,
             CheckpointLockStateChecker stateChecker,
             PersistentPageMemoryMetricSource metricSource
     ) {
+        this.logThresholdNanos = logThresholdNanos;
         this.pageMemory = pageMemory;
         this.cpProgress = cpProgress;
         this.stateChecker = stateChecker;
@@ -132,7 +137,7 @@ public TargetRatioPagesWriteThrottle(
 
             Thread curThread = Thread.currentThread();
 
-            if (throttleParkTimeNs > LOGGING_THRESHOLD) {
+            if (throttleParkTimeNs > logThresholdNanos) {
                 LOG.warn("Parking thread=" + curThread.getName()
                         + " for timeout(ms)=" + TimeUnit.NANOSECONDS.toMillis(throttleParkTimeNs));
             }
@@ -147,7 +152,7 @@ public TargetRatioPagesWriteThrottle(
                 } finally {
                     cpBufThrottledThreads.remove(curThread.getId());
 
-                    if (throttleParkTimeNs > LOGGING_THRESHOLD) {
+                    if (throttleParkTimeNs > logThresholdNanos) {
                         LOG.warn("Unparking thread=" + curThread.getName()
                                 + " with park timeout(ms)=" + TimeUnit.NANOSECONDS.toMillis(throttleParkTimeNs));
                     }

modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/throttling/ThrottlingType.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.pagememory.persistence.throttling;
+
+/**
+ * Available throttling types.
+ */
+public enum ThrottlingType {
+    /** Corresponds to no throttling. */
+    DISABLED,
+
+    /** Corresponds to {@link TargetRatioPagesWriteThrottle}. */
+    TARGET_RATIO,
+
+    /** Corresponds to {@link PagesWriteSpeedBasedThrottle}. */
+    SPEED_BASED
+}

modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/throttling/PageMemoryThrottlingTest.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.pagememory.persistence.throttling;
 
 import static org.apache.ignite.internal.configuration.ConfigurationTestUtils.fixConfiguration;
+import static org.apache.ignite.internal.pagememory.persistence.throttling.PagesWriteThrottlePolicy.DEFAULT_LOGGING_THRESHOLD;
 import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -372,13 +373,15 @@ void hugeLoadDoesNotBreakCheckpointReadLock(boolean speedBasedThrottling) {
         PagesWriteThrottlePolicy writeThrottle;
         if (speedBasedThrottling) {
             writeThrottle = new PagesWriteSpeedBasedThrottle(
+                    DEFAULT_LOGGING_THRESHOLD,
                     pageMemory,
                     checkpointManager::currentCheckpointProgress,
                     checkpointManager.checkpointTimeoutLock()::checkpointLockIsHeldByThread,
                     metricSource
             );
         } else {
             writeThrottle = new TargetRatioPagesWriteThrottle(
+                    DEFAULT_LOGGING_THRESHOLD,
                     pageMemory,
                     checkpointManager::currentCheckpointProgress,
                     checkpointManager.checkpointTimeoutLock()::checkpointLockIsHeldByThread,