ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [20/26] ignite git commit: IGNITE-6334 Throttle writing threads during ongoing checkpoint - Fixes #2710.
Date Fri, 22 Sep 2017 11:19:51 GMT
IGNITE-6334 Throttle writing threads during ongoing checkpoint - Fixes #2710.

Signed-off-by: Alexey Goncharuk <alexey.goncharuk@gmail.com>


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f1205511
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f1205511
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f1205511

Branch: refs/heads/ignite-3478
Commit: f120551159fd2eeed8cedc0ca3e3ddc394505737
Parents: 1649c53
Author: Ivan Rakov <ivan.glukos@gmail.com>
Authored: Fri Sep 22 12:40:22 2017 +0300
Committer: Alexey Goncharuk <alexey.goncharuk@gmail.com>
Committed: Fri Sep 22 12:40:22 2017 +0300

----------------------------------------------------------------------
 .../apache/ignite/IgniteSystemProperties.java   |   5 +
 .../PersistentStoreConfiguration.java           |  28 +-
 .../GridCacheDatabaseSharedManager.java         |  48 ++-
 .../persistence/pagemem/PageMemoryImpl.java     |  96 +++++-
 .../persistence/pagemem/PagesWriteThrottle.java | 104 ++++++
 .../pagemem/BPlusTreePageMemoryImplTest.java    |   4 +-
 .../BPlusTreeReuseListPageMemoryImplTest.java   |   3 +-
 .../MetadataStoragePageMemoryImplTest.java      |   4 +-
 .../pagemem/PageMemoryImplNoLoadTest.java       |   4 +-
 .../persistence/pagemem/PageMemoryImplTest.java |   4 +-
 .../pagemem/PagesWriteThrottleSandboxTest.java  | 264 +++++++++++++++
 .../pagemem/PagesWriteThrottleSmokeTest.java    | 322 +++++++++++++++++++
 .../ignite/testsuites/IgnitePdsTestSuite.java   |   4 +
 13 files changed, 866 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
index 5f1839b..39e65e1 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
@@ -24,6 +24,7 @@ import java.util.Map;
 import java.util.Properties;
 import javax.net.ssl.HostnameVerifier;
 import org.apache.ignite.cluster.ClusterGroup;
+import org.apache.ignite.configuration.PersistentStoreConfiguration;
 import org.apache.ignite.internal.marshaller.optimized.OptimizedMarshaller;
 import org.apache.ignite.lang.IgnitePredicate;
 import org.jetbrains.annotations.Nullable;
@@ -726,6 +727,10 @@ public final class IgniteSystemProperties {
      */
     public static final String IGNITE_WAL_LOG_TX_RECORDS = "IGNITE_WAL_LOG_TX_RECORDS";
 
+    /** If this property is set, {@link PersistentStoreConfiguration#writeThrottlingEnabled} will be overridden to true
+     * independent of initial value in configuration. */
+    public static final String IGNITE_OVERRIDE_WRITE_THROTTLING_ENABLED = "IGNITE_OVERRIDE_WRITE_THROTTLING_ENABLED";
+
     /**
      * Enforces singleton.
      */

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/main/java/org/apache/ignite/configuration/PersistentStoreConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/PersistentStoreConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/PersistentStoreConfiguration.java
index abca5a5..c44e92d 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/PersistentStoreConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/PersistentStoreConfiguration.java
@@ -84,6 +84,9 @@ public class PersistentStoreConfiguration implements Serializable {
     /** Default wal archive directory. */
     public static final String DFLT_WAL_ARCHIVE_PATH = "db/wal/archive";
 
+    /** Default write throttling enabled. */
+    public static final boolean DFLT_WRITE_THROTTLING_ENABLED = false;
+
     /** */
     private String persistenceStorePath;
 
@@ -162,6 +165,11 @@ public class PersistentStoreConfiguration implements Serializable {
     private long walAutoArchiveAfterInactivity = -1;
 
     /**
+     * If true, threads that generate dirty pages too fast during ongoing checkpoint will be throttled.
+     */
+    private boolean writeThrottlingEnabled = DFLT_WRITE_THROTTLING_ENABLED;
+
+    /**
      * Returns a path the root directory where the Persistent Store will persist data and indexes.
      */
     public String getPersistentStorePath() {
@@ -240,7 +248,7 @@ public class PersistentStoreConfiguration implements Serializable {
     /**
      * Sets a number of threads to use for the checkpointing purposes.
      *
-     * @param checkpointingThreads Number of checkpointing threads. One thread is used by default.
+     * @param checkpointingThreads Number of checkpointing threads. Four threads are used by default.
      * @return {@code this} for chaining.
      */
     public PersistentStoreConfiguration setCheckpointingThreads(int checkpointingThreads) {
@@ -402,6 +410,24 @@ public class PersistentStoreConfiguration implements Serializable {
     }
 
     /**
+     * Gets flag indicating whether write throttling is enabled.
+     */
+    public boolean isWriteThrottlingEnabled() {
+        return writeThrottlingEnabled;
+    }
+
+    /**
+     * Sets flag indicating whether write throttling is enabled.
+     *
+     * @param writeThrottlingEnabled Write throttling enabled flag.
+     */
+    public PersistentStoreConfiguration setWriteThrottlingEnabled(boolean writeThrottlingEnabled) {
+        this.writeThrottlingEnabled = writeThrottlingEnabled;
+
+        return this;
+    }
+
+    /**
      * Gets the length of the time interval for rate-based metrics. This interval defines a window over which
      * hits will be tracked. Default value is {@link #DFLT_RATE_TIME_INTERVAL_MILLIS}.
      *

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
index 5f03d8f..62210dc 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
@@ -301,6 +301,12 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
     /** */
     private ObjectName persistenceMetricsMbeanName;
 
+    /** Counter for written checkpoint pages. Not null only if checkpoint is running. */
+    private volatile AtomicInteger writtenPagesCntr = null;
+
+    /** Number of pages in current checkpoint. */
+    private volatile int currCheckpointPagesCnt;
+
     /**
      * @param ctx Kernal context.
      */
@@ -667,6 +673,11 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
                 "Checkpoint page buffer size is too big, setting to an adjusted cache size [size="
                     + U.readableSize(cacheSize, false) + ",  memPlc=" + plcCfg.getName() + ']');
 
+        boolean writeThrottlingEnabled = persistenceCfg.isWriteThrottlingEnabled();
+
+        if (IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_OVERRIDE_WRITE_THROTTLING_ENABLED, false))
+            writeThrottlingEnabled = true;
+
         PageMemoryImpl pageMem = new PageMemoryImpl(
             memProvider,
             calculateFragmentSizes(
@@ -699,7 +710,8 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
                 }
             },
             this,
-            memMetrics
+            memMetrics,
+            writeThrottlingEnabled
         );
 
         memMetrics.pageMemory(pageMem);
@@ -942,7 +954,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
     }
 
     /**
-     * Gets the checkpoint read lock. While this lock is held, checkpoint thread will not acquiSnapshotWorkerre memory
+     * Gets the checkpoint read lock. While this lock is held, checkpoint thread will not acquireSnapshotWorker memory
      * state.
      */
     @SuppressWarnings("LockAcquiredButNotSafelyReleased")
@@ -1906,6 +1918,20 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
     }
 
     /**
+     * Counter for written checkpoint pages. Not null only if checkpoint is running.
+     */
+    public AtomicInteger writtenPagesCounter() {
+        return writtenPagesCntr;
+    }
+
+    /**
+     * @return Number of pages in current checkpoint. If checkpoint is not running, returns 0.
+     */
+    public int currentCheckpointPagesCount() {
+        return currCheckpointPagesCnt;
+    }
+
+    /**
      * @param cpTs Checkpoint timestamp.
      * @param cpId Checkpoint ID.
      * @param type Checkpoint type.
@@ -2038,6 +2064,10 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
 
                 Checkpoint chp = markCheckpointBegin(tracker);
 
+                currCheckpointPagesCnt = chp.pagesSize;
+
+                writtenPagesCntr = new AtomicInteger();
+
                 boolean interrupted = true;
 
                 try {
@@ -2049,7 +2079,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
                             asyncRunner == null ? 1 : chp.cpPages.collectionsSize());
 
                         tracker.onPagesWriteStart();
-                        final AtomicInteger writtenPagesCtr = new AtomicInteger();
+
                         final int totalPagesToWriteCnt = chp.cpPages.size();
 
                         if (asyncRunner != null) {
@@ -2059,7 +2089,6 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
                                     chp.cpPages.innerCollection(i),
                                     updStores,
                                     doneWriteFut,
-                                    writtenPagesCtr,
                                     totalPagesToWriteCnt
                                 );
 
@@ -2078,7 +2107,6 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
                                 chp.cpPages,
                                 updStores,
                                 doneWriteFut,
-                                writtenPagesCtr,
                                 totalPagesToWriteCnt);
 
                             write.run();
@@ -2402,6 +2430,10 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
                         chp.cpEntry.checkpointMark(),
                         null,
                         CheckpointEntryType.END);
+
+                writtenPagesCntr = null;
+
+                currCheckpointPagesCnt = 0;
             }
 
             checkpointHist.onCheckpointFinished(chp);
@@ -2498,9 +2530,6 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
         /** */
         private CountDownFuture doneFut;
 
-        /** Counter for all written pages. May be shared between several workers */
-        private AtomicInteger writtenPagesCntr;
-
         /** Total pages to write, counter may be greater than {@link #writePageIds} size */
         private final int totalPagesToWrite;
 
@@ -2511,7 +2540,6 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
          * @param writePageIds Collection of page IDs to write.
          * @param updStores
          * @param doneFut
-         * @param writtenPagesCntr all written pages counter, may be shared between several write tasks
          * @param totalPagesToWrite total pages to be written under this checkpoint
          */
         private WriteCheckpointPages(
@@ -2519,13 +2547,11 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
             final Collection<FullPageId> writePageIds,
             final GridConcurrentHashSet<PageStore> updStores,
             final CountDownFuture doneFut,
-            @NotNull final AtomicInteger writtenPagesCntr,
             final int totalPagesToWrite) {
             this.tracker = tracker;
             this.writePageIds = writePageIds;
             this.updStores = updStores;
             this.doneFut = doneFut;
-            this.writtenPagesCntr = writtenPagesCntr;
             this.totalPagesToWrite = totalPagesToWrite;
         }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java
index dbb64f8..1da17b5 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java
@@ -33,6 +33,7 @@ import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
@@ -56,6 +57,7 @@ import org.apache.ignite.internal.pagemem.wal.record.delta.InitNewPageRecord;
 import org.apache.ignite.internal.pagemem.wal.record.delta.PageDeltaRecord;
 import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 import org.apache.ignite.internal.processors.cache.persistence.CheckpointLockStateChecker;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
 import org.apache.ignite.internal.processors.cache.persistence.MemoryMetricsImpl;
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.TrackingPageIO;
@@ -179,6 +181,9 @@ public class PageMemoryImpl implements PageMemoryEx {
     /** State checker. */
     private final CheckpointLockStateChecker stateChecker;
 
+    /** Number of used pages in checkpoint buffer. */
+    private final AtomicInteger cpBufPagesCntr = new AtomicInteger(0);
+
     /** */
     private ExecutorService asyncRunner = new ThreadPoolExecutor(
         0,
@@ -217,6 +222,12 @@ public class PageMemoryImpl implements PageMemoryEx {
     /** Flush dirty page closure. When possible, will be called by evictPage(). */
     private final GridInClosure3X<Long, FullPageId, PageMemoryEx> changeTracker;
 
+    /** Pages write throttle. */
+    private PagesWriteThrottle writeThrottle;
+
+    /** Write throttle enabled flag. */
+    private boolean throttleEnabled;
+
     /**  */
     private boolean pageEvictWarned;
 
@@ -232,6 +243,7 @@ public class PageMemoryImpl implements PageMemoryEx {
      * @param pageSize Page size.
      * @param flushDirtyPage Callback invoked when a dirty page is evicted.
      * @param changeTracker Callback invoked to track changes in pages.
+     * @param throttleEnabled Write throttle enabled flag.
      */
     public PageMemoryImpl(
         DirectMemoryProvider directMemoryProvider,
@@ -241,7 +253,8 @@ public class PageMemoryImpl implements PageMemoryEx {
         GridInClosure3X<FullPageId, ByteBuffer, Integer> flushDirtyPage,
         GridInClosure3X<Long, FullPageId, PageMemoryEx> changeTracker,
         CheckpointLockStateChecker stateChecker,
-        MemoryMetricsImpl memMetrics
+        MemoryMetricsImpl memMetrics,
+        boolean throttleEnabled
     ) {
         assert sharedCtx != null;
 
@@ -253,6 +266,7 @@ public class PageMemoryImpl implements PageMemoryEx {
         this.flushDirtyPage = flushDirtyPage;
         this.changeTracker = changeTracker;
         this.stateChecker = stateChecker;
+        this.throttleEnabled = throttleEnabled;
 
         storeMgr = sharedCtx.pageStore();
         walMgr = sharedCtx.wal();
@@ -290,7 +304,7 @@ public class PageMemoryImpl implements PageMemoryEx {
 
         DirectMemoryRegion cpReg = regions.get(regs - 1);
 
-        checkpointPool = new PagePool(regs - 1, cpReg);
+        checkpointPool = new PagePool(regs - 1, cpReg, cpBufPagesCntr);
 
         long checkpointBuf = cpReg.size();
 
@@ -305,12 +319,14 @@ public class PageMemoryImpl implements PageMemoryEx {
 
             totalAllocated += reg.size();
 
-            segments[i] = new Segment(i, regions.get(i), checkpointPool.pages() / segments.length);
+            segments[i] = new Segment(i, regions.get(i), checkpointPool.pages() / segments.length, throttleEnabled);
 
             pages += segments[i].pages();
             totalTblSize += segments[i].tableSize();
         }
 
+        initWriteThrottle();
+
         if (log.isInfoEnabled())
             log.info("Started page memory [memoryAllocated=" + U.readableSize(totalAllocated, false) +
                 ", pages=" + pages +
@@ -319,6 +335,21 @@ public class PageMemoryImpl implements PageMemoryEx {
                 ']');
     }
 
+    /**
+     *
+     */
+    private void initWriteThrottle() {
+        if (!(sharedCtx.database() instanceof GridCacheDatabaseSharedManager)) {
+            log.error("Write throttle can't start. Unexpected class of database manager: " +
+                sharedCtx.database().getClass());
+
+            throttleEnabled = false;
+        }
+
+        if (throttleEnabled)
+            writeThrottle = new PagesWriteThrottle(this, (GridCacheDatabaseSharedManager)sharedCtx.database());
+    }
+
     /** {@inheritDoc} */
     @SuppressWarnings("OverlyStrongTypeCast")
     @Override public void stop() throws IgniteException {
@@ -774,6 +805,18 @@ public class PageMemoryImpl implements PageMemoryEx {
         return true;
     }
 
+    /**
+     * @param dirtyRatioThreshold Throttle threshold.
+     */
+    boolean shouldThrottle(double dirtyRatioThreshold) {
+        for (Segment segment : segments) {
+            if (segment.shouldThrottle(dirtyRatioThreshold))
+                return true;
+        }
+
+        return false;
+    }
+
     /** {@inheritDoc} */
     @Override public GridMultiCollectionWrapper<FullPageId> beginCheckpoint() throws IgniteException {
         Collection[] collections = new Collection[segments.length];
@@ -799,6 +842,9 @@ public class PageMemoryImpl implements PageMemoryEx {
     @Override public void finishCheckpoint() {
         for (Segment seg : segments)
             seg.segCheckpointPages = null;
+
+        if (throttleEnabled)
+            writeThrottle.onFinishCheckpoint();
     }
 
     /** {@inheritDoc} */
@@ -1219,6 +1265,9 @@ public class PageMemoryImpl implements PageMemoryEx {
 
         try {
             rwLock.writeUnlock(page + PAGE_LOCK_OFFSET, PageIdUtils.tag(pageId));
+
+            if (throttleEnabled && !restore && markDirty && !dirty)
+                writeThrottle.onMarkDirty(isInCheckpoint(fullId));
         }
         catch (AssertionError ex) {
             StringBuilder sb = new StringBuilder(sysPageSize * 2);
@@ -1310,6 +1359,20 @@ public class PageMemoryImpl implements PageMemoryEx {
     }
 
     /**
+     * Number of used pages in checkpoint buffer.
+     */
+    public int checkpointBufferPagesCount() {
+        return cpBufPagesCntr.get();
+    }
+
+    /**
+     * Number of used pages in checkpoint buffer.
+     */
+    public int checkpointBufferPagesSize() {
+        return checkpointPool.pages();
+    }
+
+    /**
      * This method must be called in synchronized context.
      *
      * @param absPtr Absolute pointer.
@@ -1385,6 +1448,9 @@ public class PageMemoryImpl implements PageMemoryEx {
         /** Direct memory region. */
         protected final DirectMemoryRegion region;
 
+        /** Pool pages counter. */
+        protected final AtomicInteger pagesCntr;
+
         /** */
         protected long lastAllocatedIdxPtr;
 
@@ -1397,10 +1463,12 @@ public class PageMemoryImpl implements PageMemoryEx {
         /**
          * @param idx Index.
          * @param region Region
+         * @param pagesCntr Pages counter.
          */
-        protected PagePool(int idx, DirectMemoryRegion region) {
+        protected PagePool(int idx, DirectMemoryRegion region, AtomicInteger pagesCntr) {
             this.idx = idx;
             this.region = region;
+            this.pagesCntr = pagesCntr;
 
             long base = (region.address() + 7) & ~0x7;
 
@@ -1427,6 +1495,9 @@ public class PageMemoryImpl implements PageMemoryEx {
          * @throws GridOffHeapOutOfMemoryException If failed to allocate new free page.
          */
         private long borrowOrAllocateFreePage(long pageId) throws GridOffHeapOutOfMemoryException {
+            if (pagesCntr != null)
+                pagesCntr.getAndIncrement();
+
             long relPtr = borrowFreePage();
 
             return relPtr != INVALID_REL_PTR ? relPtr : allocateFreePage(pageId);
@@ -1500,6 +1571,9 @@ public class PageMemoryImpl implements PageMemoryEx {
 
             assert !PageHeader.isAcquired(absPtr) : "Release pinned page: " + PageHeader.fullPageId(absPtr);
 
+            if (pagesCntr != null)
+                pagesCntr.getAndDecrement();
+
             while (true) {
                 long freePageRelPtrMasked = GridUnsafe.getLong(freePageListPtr);
 
@@ -1580,8 +1654,9 @@ public class PageMemoryImpl implements PageMemoryEx {
 
         /**
          * @param region Memory region.
+         * @param throttlingEnabled Write throttling enabled flag.
          */
-        private Segment(int idx, DirectMemoryRegion region, int cpPoolPages) {
+        private Segment(int idx, DirectMemoryRegion region, int cpPoolPages, boolean throttlingEnabled) {
             long totalMemory = region.size();
 
             int pages = (int)(totalMemory / sysPageSize);
@@ -1596,9 +1671,9 @@ public class PageMemoryImpl implements PageMemoryEx {
 
             DirectMemoryRegion poolRegion = region.slice(memPerTbl + 8);
 
-            pool = new PagePool(idx, poolRegion);
+            pool = new PagePool(idx, poolRegion, null);
 
-            maxDirtyPages = Math.min(pool.pages() * 2 / 3, cpPoolPages);
+            maxDirtyPages = throttlingEnabled ? pool.pages() * 3 / 4 : Math.min(pool.pages() * 2 / 3, cpPoolPages);
         }
 
         /**
@@ -1609,6 +1684,13 @@ public class PageMemoryImpl implements PageMemoryEx {
         }
 
         /**
+         * @param dirtyRatioThreshold Throttle threshold.
+         */
+        private boolean shouldThrottle(double dirtyRatioThreshold) {
+            return ((double)dirtyPages.size()) / pages() > dirtyRatioThreshold;
+        }
+
+        /**
          * @return Max number of pages this segment can allocate.
          */
         private int pages() {

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottle.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottle.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottle.java
new file mode 100644
index 0000000..d0c67c7
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottle.java
@@ -0,0 +1,104 @@
+/*
+* 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.processors.cache.persistence.pagemem;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+
+/**
+ * Throttles threads that generate dirty pages during ongoing checkpoint.
+ * Designed to avoid zero dropdowns that can happen if checkpoint buffer is overflowed.
+ */
+public class PagesWriteThrottle {
+    /** Page memory. */
+    private final PageMemoryImpl pageMemory;
+
+    /** Database manager. */
+    private final GridCacheDatabaseSharedManager dbSharedMgr;
+
+    /** Starting throttle time. Limits write speed to 1000 MB/s. */
+    private static final long STARTING_THROTTLE_NANOS = 4000;
+
+    /** Backoff ratio. Each next park will be this times longer. */
+    private static final double BACKOFF_RATIO = 1.05;
+
+    /** Exponential backoff counter. */
+    private final AtomicInteger exponentialBackoffCntr = new AtomicInteger(0);
+    /**
+     * @param pageMemory Page memory.
+     * @param dbSharedMgr Database manager.
+     */
+    public PagesWriteThrottle(PageMemoryImpl pageMemory, GridCacheDatabaseSharedManager dbSharedMgr) {
+        this.pageMemory = pageMemory;
+        this.dbSharedMgr = dbSharedMgr;
+    }
+
+    /**
+     *
+     */
+    public void onMarkDirty(boolean isInCheckpoint) {
+        assert dbSharedMgr.checkpointLockIsHeldByThread();
+
+        AtomicInteger writtenPagesCntr = dbSharedMgr.writtenPagesCounter();
+
+        if (writtenPagesCntr == null)
+            return; // Don't throttle if checkpoint is not running.
+
+        boolean shouldThrottle = false;
+
+        if (isInCheckpoint) {
+            int checkpointBufLimit = pageMemory.checkpointBufferPagesSize() * 2 / 3;
+
+            shouldThrottle = pageMemory.checkpointBufferPagesCount() > checkpointBufLimit;
+        }
+
+        if (!shouldThrottle) {
+            int cpWrittenPages = writtenPagesCntr.get();
+
+            int cpTotalPages = dbSharedMgr.currentCheckpointPagesCount();
+
+            if (cpWrittenPages == cpTotalPages) {
+                // Checkpoint is already in fsync stage, increasing maximum ratio of dirty pages to 3/4
+                shouldThrottle = pageMemory.shouldThrottle(3.0 / 4);
+            } else {
+                double dirtyRatioThreshold = ((double)cpWrittenPages) / cpTotalPages;
+
+                // Starting with 0.05 to avoid throttle right after checkpoint start
+                // 7/12 is maximum ratio of dirty pages
+                dirtyRatioThreshold = (dirtyRatioThreshold * 0.95 + 0.05) * 7 / 12;
+
+                shouldThrottle = pageMemory.shouldThrottle(dirtyRatioThreshold);
+            }
+        }
+
+        if (shouldThrottle) {
+            int throttleLevel = exponentialBackoffCntr.getAndIncrement();
+
+            LockSupport.parkNanos((long)(STARTING_THROTTLE_NANOS * Math.pow(BACKOFF_RATIO, throttleLevel)));
+        }
+        else
+            exponentialBackoffCntr.set(0);
+    }
+
+    /**
+     *
+     */
+    public void onFinishCheckpoint() {
+        exponentialBackoffCntr.set(0);
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
index 6f58782..56d09f8 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
@@ -82,7 +82,9 @@ public class BPlusTreePageMemoryImplTest extends BPlusTreeSelfTest {
                     return true;
                 }
             },
-            new MemoryMetricsImpl(new MemoryPolicyConfiguration()));
+            new MemoryMetricsImpl(new MemoryPolicyConfiguration()),
+            false
+        );
 
         mem.start();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
index b263d4f..39183b2 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
@@ -82,7 +82,8 @@ public class BPlusTreeReuseListPageMemoryImplTest extends BPlusTreeReuseSelfTest
                     return true;
                 }
             },
-            new MemoryMetricsImpl(new MemoryPolicyConfiguration())
+            new MemoryMetricsImpl(new MemoryPolicyConfiguration()),
+            false
         );
 
         mem.start();

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/MetadataStoragePageMemoryImplTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/MetadataStoragePageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/MetadataStoragePageMemoryImplTest.java
index d9257bd..a427c63 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/MetadataStoragePageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/MetadataStoragePageMemoryImplTest.java
@@ -97,6 +97,8 @@ public class MetadataStoragePageMemoryImplTest extends MetadataStorageSelfTest{
                     return true;
                 }
             },
-            new MemoryMetricsImpl(new MemoryPolicyConfiguration()));
+            new MemoryMetricsImpl(new MemoryPolicyConfiguration()),
+            false
+        );
     }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
index 1fff1f0..467ede4 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
@@ -88,7 +88,9 @@ public class PageMemoryImplNoLoadTest extends PageMemoryNoLoadSelfTest {
                     return true;
                 }
             },
-            new MemoryMetricsImpl(new MemoryPolicyConfiguration()));
+            new MemoryMetricsImpl(new MemoryPolicyConfiguration()),
+            false
+        );
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
index 0366eca..c5997fa 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
@@ -110,7 +110,9 @@ public class PageMemoryImplTest extends GridCommonAbstractTest {
                     return true;
                 }
             },
-            new MemoryMetricsImpl(new MemoryPolicyConfiguration()));
+            new MemoryMetricsImpl(new MemoryPolicyConfiguration()),
+            false
+        );
 
         mem.start();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottleSandboxTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottleSandboxTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottleSandboxTest.java
new file mode 100644
index 0000000..409ab84
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottleSandboxTest.java
@@ -0,0 +1,264 @@
+/*
+* 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.processors.cache.persistence.pagemem;
+
+import java.io.Serializable;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.MemoryMetrics;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.MemoryConfiguration;
+import org.apache.ignite.configuration.MemoryPolicyConfiguration;
+import org.apache.ignite.configuration.PersistentStoreConfiguration;
+import org.apache.ignite.configuration.WALMode;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.ratemetrics.HitRateMetrics;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ * Test to visualize and debug {@link PagesWriteThrottle}.
+ * Prints puts/gets rate, number of dirty pages, pages written in current checkpoint and pages in checkpoint buffer.
+ * Not intended to be part of any test suite.
+ */
+public class PagesWriteThrottleSandboxTest extends GridCommonAbstractTest {
+    /** Ip finder. */
+    private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+    /** Cache name. */
+    private static final String CACHE_NAME = "cache1";
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        TcpDiscoverySpi discoverySpi = (TcpDiscoverySpi)cfg.getDiscoverySpi();
+        discoverySpi.setIpFinder(ipFinder);
+
+        MemoryConfiguration dbCfg = new MemoryConfiguration();
+
+        dbCfg.setMemoryPolicies(new MemoryPolicyConfiguration()
+            .setMaxSize(4000L * 1024 * 1024)
+            .setName("dfltMemPlc")
+            .setMetricsEnabled(true));
+
+        dbCfg.setDefaultMemoryPolicyName("dfltMemPlc");
+
+        cfg.setMemoryConfiguration(dbCfg);
+
+        CacheConfiguration ccfg1 = new CacheConfiguration();
+
+        ccfg1.setName(CACHE_NAME);
+        ccfg1.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
+        ccfg1.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+        ccfg1.setAffinity(new RendezvousAffinityFunction(false, 64));
+
+        cfg.setCacheConfiguration(ccfg1);
+
+        cfg.setPersistentStoreConfiguration(
+            new PersistentStoreConfiguration()
+                .setWalMode(WALMode.BACKGROUND)
+                .setCheckpointingFrequency(20_000)
+                .setCheckpointingPageBufferSize(1000L * 1000 * 1000)
+                .setWriteThrottlingEnabled(true));
+
+        cfg.setConsistentId(gridName);
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        super.beforeTest();
+
+        deleteWorkFiles();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        super.afterTest();
+
+        deleteWorkFiles();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected long getTestTimeout() {
+        return 100 * 60 * 1000;
+    }
+
+    /**
+     * @throws Exception if failed.
+     */
+    public void testThrottle() throws Exception {
+        startGrids(1).active(true);
+
+        try {
+            final Ignite ig = ignite(0);
+
+            final int keyCnt = 4_000_000;
+
+            final AtomicBoolean run = new AtomicBoolean(true);
+
+            final HitRateMetrics getRate = new HitRateMetrics(5000, 5);
+
+            GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
+                @Override public Object call() throws Exception {
+                    while (run.get()) {
+                        ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+                        int key = rnd.nextInt(keyCnt * 2);
+
+                        ignite(0).cache(CACHE_NAME).get(key);
+
+                        getRate.onHit();
+                    }
+
+                    return null;
+                }
+            }, 2, "read-loader");
+
+            final HitRateMetrics putRate = new HitRateMetrics(1000, 5);
+
+            GridTestUtils.runAsync(new Runnable() {
+                @Override public void run() {
+                    while (run.get()) {
+                        long dirtyPages = 0;
+
+                        for (MemoryMetrics m : ig.memoryMetrics())
+                            if (m.getName().equals("dfltMemPlc"))
+                                dirtyPages = m.getDirtyPages();
+
+                        long cpBufPages = 0;
+
+                        long cpWrittenPages;
+
+                        AtomicInteger cntr = ((GridCacheDatabaseSharedManager)(((IgniteEx)ignite(0))
+                            .context().cache().context().database())).writtenPagesCounter();
+
+                        cpWrittenPages = cntr == null ? 0 : cntr.get();
+
+                        try {
+                            cpBufPages = ((PageMemoryImpl)((IgniteEx)ignite(0)).context().cache().context().database()
+                                .memoryPolicy("dfltMemPlc").pageMemory()).checkpointBufferPagesCount();
+                        }
+                        catch (IgniteCheckedException e) {
+                            e.printStackTrace();
+                        }
+
+                        System.out.println("@@@ putsPerSec=," + (putRate.getRate()) + ", getsPerSec=," + (getRate.getRate())  + ", dirtyPages=," + dirtyPages + ", cpWrittenPages=," + cpWrittenPages +", cpBufPages=," + cpBufPages);
+
+                        try {
+                            Thread.sleep(1000);
+                        }
+                        catch (InterruptedException e) {
+                            Thread.currentThread().interrupt();
+                        }
+                    }
+                }
+            }, "metrics-view");
+
+            try (IgniteDataStreamer<Object, Object> ds = ig.dataStreamer(CACHE_NAME)) {
+                ds.allowOverwrite(true);
+
+                for (int i = 0; i < keyCnt * 10; i++) {
+                    ds.addData(ThreadLocalRandom.current().nextInt(keyCnt), new TestValue(ThreadLocalRandom.current().nextInt(),
+                        ThreadLocalRandom.current().nextInt()));
+
+                    putRate.onHit();
+                }
+            }
+
+            run.set(false);
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     *
+     */
+    private static class TestValue implements Serializable {
+        /** */
+        private final int v1;
+
+        /** */
+        private final int v2;
+
+        /** */
+        private byte[] payload = new byte[400 + ThreadLocalRandom.current().nextInt(20)];
+
+        /**
+         * @param v1 Value 1.
+         * @param v2 Value 2.
+         */
+        private TestValue(int v1, int v2) {
+            this.v1 = v1;
+            this.v2 = v2;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean equals(Object o) {
+            if (this == o)
+                return true;
+
+            if (o == null || getClass() != o.getClass())
+                return false;
+
+            TestValue val = (TestValue)o;
+
+            return v1 == val.v1 && v2 == val.v2;
+        }
+
+        /** {@inheritDoc} */
+        @Override public int hashCode() {
+            int res = v1;
+
+            res = 31 * res + v2;
+
+            return res;
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(TestValue.class, this);
+        }
+    }
+
+    /**
+     * @throws IgniteCheckedException If failed.
+     */
+    private void deleteWorkFiles() throws IgniteCheckedException {
+        deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+        deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "snapshot", false));
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottleSmokeTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottleSmokeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottleSmokeTest.java
new file mode 100644
index 0000000..12a601d
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottleSmokeTest.java
@@ -0,0 +1,322 @@
+/*
+* 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.processors.cache.persistence.pagemem;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.LockSupport;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.MemoryConfiguration;
+import org.apache.ignite.configuration.MemoryPolicyConfiguration;
+import org.apache.ignite.configuration.PersistentStoreConfiguration;
+import org.apache.ignite.configuration.WALMode;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
+import org.apache.ignite.internal.processors.cache.persistence.file.FileIODecorator;
+import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
+import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory;
+import org.apache.ignite.internal.processors.cache.ratemetrics.HitRateMetrics;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class PagesWriteThrottleSmokeTest extends GridCommonAbstractTest {
+    /** Ip finder. */
+    private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+    /** Slow checkpoint enabled. */
+    private final AtomicBoolean slowCheckpointEnabled = new AtomicBoolean(true);
+
+    /** Cache name. */
+    private static final String CACHE_NAME = "cache1";
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        TcpDiscoverySpi discoverySpi = (TcpDiscoverySpi)cfg.getDiscoverySpi();
+        discoverySpi.setIpFinder(ipFinder);
+
+        MemoryConfiguration dbCfg = new MemoryConfiguration();
+
+        dbCfg.setMemoryPolicies(new MemoryPolicyConfiguration()
+            .setMaxSize(400 * 1024 * 1024)
+            .setName("dfltMemPlc")
+            .setMetricsEnabled(true));
+
+        dbCfg.setDefaultMemoryPolicyName("dfltMemPlc");
+
+        cfg.setMemoryConfiguration(dbCfg);
+
+        CacheConfiguration ccfg1 = new CacheConfiguration();
+
+        ccfg1.setName(CACHE_NAME);
+        ccfg1.setAtomicityMode(CacheAtomicityMode.ATOMIC);
+        ccfg1.setWriteSynchronizationMode(CacheWriteSynchronizationMode.PRIMARY_SYNC);
+        ccfg1.setAffinity(new RendezvousAffinityFunction(false, 64));
+
+        cfg.setCacheConfiguration(ccfg1);
+
+        cfg.setPersistentStoreConfiguration(
+            new PersistentStoreConfiguration()
+                .setWalMode(WALMode.BACKGROUND)
+                .setCheckpointingFrequency(20_000)
+                .setCheckpointingPageBufferSize(200 * 1000 * 1000)
+                .setWriteThrottlingEnabled(true)
+                .setCheckpointingThreads(1)
+                .setFileIOFactory(new SlowCheckpointFileIOFactory()));
+
+        cfg.setConsistentId(gridName);
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        super.beforeTest();
+
+        deleteWorkFiles();
+
+        slowCheckpointEnabled.set(true);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        super.afterTest();
+
+        deleteWorkFiles();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected long getTestTimeout() {
+        return 6 * 60 * 1000;
+    }
+
+    /**
+     * @throws Exception if failed.
+     */
+    public void testThrottle() throws Exception {
+        startGrids(2).active(true);
+
+        try {
+            Ignite ig = ignite(0);
+
+            final int keyCnt = 2_000_000;
+
+            final AtomicBoolean run = new AtomicBoolean(true);
+
+            final AtomicBoolean zeroDropdown = new AtomicBoolean(false);
+
+            final HitRateMetrics putRate10secs = new HitRateMetrics(10_000, 20);
+
+            final HitRateMetrics putRate1sec = new HitRateMetrics(1_000, 20);
+
+            GridTestUtils.runAsync(new Runnable() {
+                @Override public void run() {
+                    try {
+                        Thread.sleep(5000);
+
+                        while (run.get()) {
+                            System.out.println(
+                                "Put rate over last 10 seconds: " + (putRate10secs.getRate() / 10) +
+                                    " puts/sec, over last 1 second: " + putRate1sec.getRate());
+
+                            if (putRate10secs.getRate() == 0) {
+                                zeroDropdown.set(true);
+
+                                run.set(false);
+                            }
+
+                            Thread.sleep(1000);
+                        }
+                    }
+                    catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    }
+                    finally {
+                        run.set(false);
+                    }
+                }
+            }, "rate-checker");
+
+            final IgniteCache<Integer, TestValue> cache = ig.getOrCreateCache(CACHE_NAME);
+
+            GridTestUtils.runAsync(new Runnable() {
+                @Override public void run() {
+                    long startTs = System.currentTimeMillis();
+
+                    for (int i = 0; i < keyCnt * 10 && System.currentTimeMillis() - startTs < 3 * 60 * 1000; i++) {
+                        if (!run.get())
+                            break;
+
+                        cache.put(ThreadLocalRandom.current().nextInt(keyCnt), new TestValue(ThreadLocalRandom.current().nextInt(),
+                            ThreadLocalRandom.current().nextInt()));
+
+                        putRate10secs.onHit();
+
+                        putRate1sec.onHit();
+                    }
+
+                    run.set(false);
+                }
+            }, "loader");
+
+            while (run.get())
+                LockSupport.parkNanos(10_000);
+
+            if (zeroDropdown.get()) {
+                slowCheckpointEnabled.set(false);
+
+                IgniteInternalFuture cpFut1 = ((IgniteEx)ignite(0)).context().cache().context().database()
+                    .wakeupForCheckpoint("test");
+
+                IgniteInternalFuture cpFut2 = ((IgniteEx)ignite(1)).context().cache().context().database()
+                    .wakeupForCheckpoint("test");
+
+                cpFut1.get();
+
+                cpFut2.get();
+
+                fail("Put rate degraded to zero for at least 10 seconds");
+            }
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     *
+     */
+    private static class TestValue implements Serializable {
+        /** */
+        private final int v1;
+
+        /** */
+        private final int v2;
+
+        /** */
+        private byte[] payload = new byte[400 + ThreadLocalRandom.current().nextInt(20)];
+
+        /**
+         * @param v1 Value 1.
+         * @param v2 Value 2.
+         */
+        private TestValue(int v1, int v2) {
+            this.v1 = v1;
+            this.v2 = v2;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean equals(Object o) {
+            if (this == o)
+                return true;
+
+            if (o == null || getClass() != o.getClass())
+                return false;
+
+            TestValue val = (TestValue)o;
+
+            return v1 == val.v1 && v2 == val.v2;
+        }
+
+        /** {@inheritDoc} */
+        @Override public int hashCode() {
+            int res = v1;
+
+            res = 31 * res + v2;
+
+            return res;
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(TestValue.class, this);
+        }
+    }
+
+    /**
+     * @throws IgniteCheckedException If failed.
+     */
+    private void deleteWorkFiles() throws IgniteCheckedException {
+        deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "db", false));
+        deleteRecursively(U.resolveWorkDirectory(U.defaultWorkDirectory(), "snapshot", false));
+    }
+
+    /**
+     * Create File I/O that emulates poor checkpoint write speed.
+     */
+    private class SlowCheckpointFileIOFactory implements FileIOFactory {
+        /** Serial version uid. */
+        private static final long serialVersionUID = 0L;
+
+        /** Delegate factory. */
+        private final FileIOFactory delegateFactory = new RandomAccessFileIOFactory();
+
+        /** {@inheritDoc} */
+        @Override public FileIO create(File file) throws IOException {
+            return create(file, "rw");
+        }
+
+        /** {@inheritDoc} */
+        @Override public FileIO create(File file, String mode) throws IOException {
+            final FileIO delegate = delegateFactory.create(file, mode);
+
+            return new FileIODecorator(delegate) {
+                @Override public int write(ByteBuffer srcBuf) throws IOException {
+                    if (slowCheckpointEnabled.get() && Thread.currentThread().getName().contains("checkpoint"))
+                        LockSupport.parkNanos(5_000_000);
+
+                    return delegate.write(srcBuf);
+                }
+
+                @Override public int write(ByteBuffer srcBuf, long position) throws IOException {
+                    if (slowCheckpointEnabled.get() && Thread.currentThread().getName().contains("checkpoint"))
+                        LockSupport.parkNanos(5_000_000);
+
+                    return delegate.write(srcBuf, position);
+                }
+
+                @Override public void write(byte[] buf, int off, int len) throws IOException {
+                    if (slowCheckpointEnabled.get() && Thread.currentThread().getName().contains("checkpoint"))
+                        LockSupport.parkNanos(5_000_000);
+
+                    delegate.write(buf, off, len);
+                }
+            };
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/f1205511/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite.java
index b2a1f65..ef7682f 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite.java
@@ -31,6 +31,7 @@ import org.apache.ignite.internal.processors.cache.persistence.pagemem.BPlusTree
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.MetadataStoragePageMemoryImplTest;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImplNoLoadTest;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImplTest;
+import org.apache.ignite.internal.processors.cache.persistence.pagemem.PagesWriteThrottleSmokeTest;
 import org.apache.ignite.internal.processors.database.IgniteDbClientNearCachePutGetTest;
 import org.apache.ignite.internal.processors.database.IgniteDbDynamicCacheSelfTest;
 import org.apache.ignite.internal.processors.database.IgniteDbMultiNodePutGetTest;
@@ -80,6 +81,9 @@ public class IgnitePdsTestSuite extends TestSuite {
 
         suite.addTestSuite(DefaultPageSizeBackwardsCompatibilityTest.class);
 
+        // Write throttling
+        suite.addTestSuite(PagesWriteThrottleSmokeTest.class);
+
         return suite;
     }
 }


Mime
View raw message