ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ira...@apache.org
Subject ignite git commit: IGNITE-8955 Checkpoint can't get write lock if massive eviction on node start started
Date Wed, 11 Jul 2018 16:43:31 GMT
Repository: ignite
Updated Branches:
  refs/heads/master 6ad291d22 -> a0fa79a8d


IGNITE-8955 Checkpoint can't get write lock if massive eviction on node start started

Signed-off-by: Ivan Rakov <irakov@apache.org>


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

Branch: refs/heads/master
Commit: a0fa79a8d297654973ec8bb890485e8e8818594c
Parents: 6ad291d
Author: Eduard Shangareev <eshangareev@gridgain.com>
Authored: Wed Jul 11 19:43:19 2018 +0300
Committer: Ivan Rakov <irakov@apache.org>
Committed: Wed Jul 11 19:43:19 2018 +0300

----------------------------------------------------------------------
 .../IgniteAuthenticationProcessor.java          |   7 +-
 .../dht/preloader/GridDhtPartitionDemander.java | 107 +++---
 .../GridCacheDatabaseSharedManager.java         | 169 +++++----
 .../persistence/GridCacheOffheapManager.java    |   1 +
 .../persistence/pagemem/PageMemoryImpl.java     |  18 +-
 .../pagemem/PagesWriteSpeedBasedThrottle.java   |  16 +-
 .../persistence/pagemem/PagesWriteThrottle.java |  19 +-
 .../pagemem/PagesWriteThrottlePolicy.java       |   5 +
 .../db/CheckpointBufferDeadlockTest.java        | 358 +++++++++++++++++++
 9 files changed, 577 insertions(+), 123 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/a0fa79a8/modules/core/src/main/java/org/apache/ignite/internal/processors/authentication/IgniteAuthenticationProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/authentication/IgniteAuthenticationProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/authentication/IgniteAuthenticationProcessor.java
index ac713c3..ded37e7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/authentication/IgniteAuthenticationProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/authentication/IgniteAuthenticationProcessor.java
@@ -1292,9 +1292,10 @@ public class IgniteAuthenticationProcessor extends GridProcessorAdapter
implemen
                 // Remove failed operation from active operations.
                 activeOps.remove(op.id());
             }
-
-            if (sharedCtx != null)
-                sharedCtx.database().checkpointReadUnlock();
+            finally {
+                if (sharedCtx != null)
+                    sharedCtx.database().checkpointReadUnlock();
+            }
 
             curOpFinishMsg = msg0;
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/a0fa79a8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
index 3cfc25f..1eeebae 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
@@ -727,77 +727,86 @@ public class GridDhtPartitionDemander {
         try {
             AffinityAssignment aff = grp.affinity().cachedAffinity(topVer);
 
-            ctx.database().checkpointReadLock();
+            // Preload.
+            for (Map.Entry<Integer, CacheEntryInfoCollection> e : supply.infos().entrySet())
{
+                int p = e.getKey();
 
-            try {
-                // Preload.
-                for (Map.Entry<Integer, CacheEntryInfoCollection> e : supply.infos().entrySet())
{
-                    int p = e.getKey();
+                if (aff.get(p).contains(ctx.localNode())) {
+                    GridDhtLocalPartition part = top.localPartition(p, topVer, true);
 
-                    if (aff.get(p).contains(ctx.localNode())) {
-                        GridDhtLocalPartition part = top.localPartition(p, topVer, true);
+                    assert part != null;
 
-                        assert part != null;
+                    boolean last = supply.last().containsKey(p);
 
-                        boolean last = supply.last().containsKey(p);
+                    if (part.state() == MOVING) {
+                        boolean reserved = part.reserve();
 
-                        if (part.state() == MOVING) {
-                            boolean reserved = part.reserve();
+                        assert reserved : "Failed to reserve partition [igniteInstanceName="
+
+                            ctx.igniteInstanceName() + ", grp=" + grp.cacheOrGroupName()
+ ", part=" + part + ']';
 
-                            assert reserved : "Failed to reserve partition [igniteInstanceName="
+
-                                ctx.igniteInstanceName() + ", grp=" + grp.cacheOrGroupName()
+ ", part=" + part + ']';
+                        part.lock();
 
-                            part.lock();
+                        try {
+                            Iterator<GridCacheEntryInfo> infos = e.getValue().infos().iterator();
 
-                            try {
-                                // Loop through all received entries and try to preload them.
-                                for (GridCacheEntryInfo entry : e.getValue().infos()) {
-                                    if (!preloadEntry(node, p, entry, topVer)) {
-                                        if (log.isDebugEnabled())
-                                            log.debug("Got entries for invalid partition
during " +
-                                                "preloading (will skip) [p=" + p + ", entry="
+ entry + ']');
-
-                                        break;
-                                    }
+                            // Loop through all received entries and try to preload them.
+                            while (infos.hasNext()) {
+                                ctx.database().checkpointReadLock();
 
-                                    for (GridCacheContext cctx : grp.caches()) {
-                                        if (cctx.statisticsEnabled())
-                                            cctx.cache().metrics0().onRebalanceKeyReceived();
-                                    }
-                                }
+                                try {
+                                    for (int i = 0; i < 100; i++) {
+                                        if (!infos.hasNext())
+                                            break;
+
+                                        GridCacheEntryInfo entry = infos.next();
 
-                                // If message was last for this partition,
-                                // then we take ownership.
-                                if (last) {
-                                    fut.partitionDone(nodeId, p, true);
+                                        if (!preloadEntry(node, p, entry, topVer)) {
+                                            if (log.isDebugEnabled())
+                                                log.debug("Got entries for invalid partition
during " +
+                                                        "preloading (will skip) [p=" + p
+ ", entry=" + entry + ']');
 
-                                    if (log.isDebugEnabled())
-                                        log.debug("Finished rebalancing partition: " + part);
+                                            break;
+                                        }
+
+                                        for (GridCacheContext cctx : grp.caches()) {
+                                            if (cctx.statisticsEnabled())
+                                                cctx.cache().metrics0().onRebalanceKeyReceived();
+                                        }
+                                    }
+                                }
+                                finally {
+                                    ctx.database().checkpointReadUnlock();
                                 }
                             }
-                            finally {
-                                part.unlock();
-                                part.release();
+
+                            // If message was last for this partition,
+                            // then we take ownership.
+                            if (last) {
+                                fut.partitionDone(nodeId, p, true);
+
+                                if (log.isDebugEnabled())
+                                    log.debug("Finished rebalancing partition: " + part);
                             }
                         }
-                        else {
-                            if (last)
-                                fut.partitionDone(nodeId, p, false);
-
-                            if (log.isDebugEnabled())
-                                log.debug("Skipping rebalancing partition (state is not MOVING):
" + part);
+                        finally {
+                            part.unlock();
+                            part.release();
                         }
                     }
                     else {
-                        fut.partitionDone(nodeId, p, false);
+                        if (last)
+                            fut.partitionDone(nodeId, p, false);
 
                         if (log.isDebugEnabled())
-                            log.debug("Skipping rebalancing partition (it does not belong
on current node): " + p);
+                            log.debug("Skipping rebalancing partition (state is not MOVING):
" + part);
                     }
                 }
-            }
-            finally {
-                ctx.database().checkpointReadUnlock();
+                else {
+                    fut.partitionDone(nodeId, p, false);
+
+                    if (log.isDebugEnabled())
+                        log.debug("Skipping rebalancing partition (it does not belong on
current node): " + p);
+                }
             }
 
             // Only request partitions based on latest topology version.

http://git-wip-us.apache.org/repos/asf/ignite/blob/a0fa79a8/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 6389942..2a0ba44 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
@@ -1478,7 +1478,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
                 throw new IgniteException(new NodeStoppingException("Failed to perform cache
update: node is stopping."));
             }
 
-            if (safeToUpdatePageMemories() || checkpointLock.getReadHoldCount() > 1)
+            if (checkpointLock.getReadHoldCount() > 1 || safeToUpdatePageMemories())
                 break;
             else {
                 checkpointLock.readLock().unlock();
@@ -2542,6 +2542,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
 
                 Integer tag = pageMem.getForCheckpoint(fullId, tmpWriteBuf, null);
 
+                assert tag == null || tag != PageMemoryImpl.TRY_AGAIN_TAG :
+                        "Lock is held by other thread for page " + fullId;
+
                 if (tag != null) {
                     tmpWriteBuf.rewind();
 
@@ -3103,7 +3106,8 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
                                     chp.cpPages.innerCollection(i),
                                     updStores,
                                     doneWriteFut,
-                                    totalPagesToWriteCnt
+                                    totalPagesToWriteCnt,
+                                    asyncRunner
                                 );
 
                                 try {
@@ -3117,11 +3121,13 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
                         }
                         else {
                             // Single-threaded checkpoint.
-                            Runnable write = new WriteCheckpointPages(tracker,
+                            Runnable write = new WriteCheckpointPages(
+                                tracker,
                                 chp.cpPages,
                                 updStores,
                                 doneWriteFut,
-                                totalPagesToWriteCnt);
+                                totalPagesToWriteCnt,
+                                null);
 
                             write.run();
                         }
@@ -3752,117 +3758,158 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
     /** Pages write task */
     private class WriteCheckpointPages implements Runnable {
         /** */
-        private CheckpointMetricsTracker tracker;
+        private final CheckpointMetricsTracker tracker;
 
         /** Collection of page IDs to write under this task. Overall pages to write may be
greater than this collection */
-        private Collection<FullPageId> writePageIds;
+        private final Collection<FullPageId> writePageIds;
 
         /** */
-        private ConcurrentLinkedHashMap<PageStore, LongAdder> updStores;
+        private final ConcurrentLinkedHashMap<PageStore, LongAdder> updStores;
 
         /** */
-        private CountDownFuture doneFut;
+        private final CountDownFuture doneFut;
 
         /** Total pages to write, counter may be greater than {@link #writePageIds} size
*/
         private final int totalPagesToWrite;
 
+        /** If any pages were skipped, new task with remaining pages will be submitted here.
*/
+        private final ExecutorService retryWriteExecutor;
+
         /**
-         * Creates task for write pages
-         *
-         * @param tracker
-         * @param writePageIds Collection of page IDs to write.
-         * @param updStores
-         * @param doneFut
-         * @param totalPagesToWrite total pages to be written under this checkpoint
+         * @param tracker Tracker.
+         * @param writePageIds Write page ids.
+         * @param updStores Upd stores.
+         * @param doneFut Done future.
+         * @param totalPagesToWrite Total pages to write.
+         * @param retryWriteExecutor Retry write executor.
          */
         private WriteCheckpointPages(
             final CheckpointMetricsTracker tracker,
             final Collection<FullPageId> writePageIds,
             final ConcurrentLinkedHashMap<PageStore, LongAdder> updStores,
             final CountDownFuture doneFut,
-            final int totalPagesToWrite) {
+            final int totalPagesToWrite,
+            final ExecutorService retryWriteExecutor
+        ) {
             this.tracker = tracker;
             this.writePageIds = writePageIds;
             this.updStores = updStores;
             this.doneFut = doneFut;
             this.totalPagesToWrite = totalPagesToWrite;
+            this.retryWriteExecutor = retryWriteExecutor;
         }
 
         /** {@inheritDoc} */
         @Override public void run() {
+            snapshotMgr.beforeCheckpointPageWritten();
+
+            Collection<FullPageId> writePageIds = this.writePageIds;
+
+            try {
+                List<FullPageId> pagesToRetry = writePages(writePageIds);
+
+                if (pagesToRetry.isEmpty())
+                    doneFut.onDone((Void)null);
+                else {
+                    if (retryWriteExecutor == null) {
+                        while (!pagesToRetry.isEmpty())
+                            pagesToRetry = writePages(pagesToRetry);
+
+                        doneFut.onDone((Void)null);
+                    }
+                    else {
+                        // Submit current retry pages to the end of the queue to avoid starvation.
+                        WriteCheckpointPages retryWritesTask = new WriteCheckpointPages(
+                            tracker, pagesToRetry, updStores, doneFut, totalPagesToWrite,
retryWriteExecutor);
+
+                        retryWriteExecutor.submit(retryWritesTask);
+                    }
+                }
+            }
+            catch (Throwable e) {
+                doneFut.onDone(e);
+            }
+        }
+
+        /**
+         * @param writePageIds Collections of pages to write.
+         * @return pagesToRetry Pages which should be retried.
+         */
+        private List<FullPageId> writePages(Collection<FullPageId> writePageIds)
throws IgniteCheckedException {
             ByteBuffer tmpWriteBuf = threadBuf.get();
 
             long writeAddr = GridUnsafe.bufferAddress(tmpWriteBuf);
 
-            snapshotMgr.beforeCheckpointPageWritten();
+            List<FullPageId> pagesToRetry = new ArrayList<>();
 
-            try {
-                for (FullPageId fullId : writePageIds) {
-                    if (checkpointer.shutdownNow)
-                        break;
-
-                    tmpWriteBuf.rewind();
+            for (FullPageId fullId : writePageIds) {
+                if (checkpointer.shutdownNow)
+                    break;
 
-                    snapshotMgr.beforePageWrite(fullId);
+                tmpWriteBuf.rewind();
 
-                    int grpId = fullId.groupId();
+                snapshotMgr.beforePageWrite(fullId);
 
-                    PageMemoryEx pageMem;
+                int grpId = fullId.groupId();
 
-                    if (grpId != MetaStorage.METASTORAGE_CACHE_ID) {
-                        CacheGroupContext grp = context().cache().cacheGroup(grpId);
+                PageMemoryEx pageMem;
 
-                        if (grp == null)
-                            continue;
+                if (grpId != MetaStorage.METASTORAGE_CACHE_ID) {
+                    CacheGroupContext grp = context().cache().cacheGroup(grpId);
 
-                        if (!grp.dataRegion().config().isPersistenceEnabled())
-                            continue;
+                    if (grp == null)
+                        continue;
 
-                        pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
-                    }
-                    else
-                        pageMem = (PageMemoryEx)metaStorage.pageMemory();
+                    if (!grp.dataRegion().config().isPersistenceEnabled())
+                        continue;
 
+                    pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
+                }
+                else
+                    pageMem = (PageMemoryEx)metaStorage.pageMemory();
 
-                    Integer tag = pageMem.getForCheckpoint(
-                        fullId, tmpWriteBuf, persStoreMetrics.metricsEnabled() ? tracker
: null);
 
-                    if (tag != null) {
-                        assert PageIO.getType(tmpWriteBuf) != 0 : "Invalid state. Type is
0! pageId = " + U.hexLong(fullId.pageId());
-                        assert PageIO.getVersion(tmpWriteBuf) != 0 : "Invalid state. Version
is 0! pageId = " + U.hexLong(fullId.pageId());
+                Integer tag = pageMem.getForCheckpoint(
+                    fullId, tmpWriteBuf, persStoreMetrics.metricsEnabled() ? tracker : null);
 
-                        tmpWriteBuf.rewind();
+                if (tag != null) {
+                    if (tag == PageMemoryImpl.TRY_AGAIN_TAG) {
+                        pagesToRetry.add(fullId);
 
-                        if (persStoreMetrics.metricsEnabled()) {
-                            int pageType = PageIO.getType(tmpWriteBuf);
+                        continue;
+                    }
 
-                            if (PageIO.isDataPageType(pageType))
-                                tracker.onDataPageWritten();
-                        }
+                    assert PageIO.getType(tmpWriteBuf) != 0 : "Invalid state. Type is 0!
pageId = " + U.hexLong(fullId.pageId());
+                    assert PageIO.getVersion(tmpWriteBuf) != 0 : "Invalid state. Version
is 0! pageId = " + U.hexLong(fullId.pageId());
 
-                        if (!skipCrc) {
-                            PageIO.setCrc(writeAddr, PureJavaCrc32.calcCrc32(tmpWriteBuf,
pageSize()));
+                    tmpWriteBuf.rewind();
 
-                            tmpWriteBuf.rewind();
-                        }
+                    if (persStoreMetrics.metricsEnabled()) {
+                        int pageType = PageIO.getType(tmpWriteBuf);
 
-                        int curWrittenPages = writtenPagesCntr.incrementAndGet();
+                        if (PageIO.isDataPageType(pageType))
+                            tracker.onDataPageWritten();
+                    }
 
-                        snapshotMgr.onPageWrite(fullId, tmpWriteBuf, curWrittenPages, totalPagesToWrite);
+                    if (!skipCrc) {
+                        PageIO.setCrc(writeAddr, PureJavaCrc32.calcCrc32(tmpWriteBuf, pageSize()));
 
                         tmpWriteBuf.rewind();
+                    }
 
-                        PageStore store = storeMgr.writeInternal(grpId, fullId.pageId(),
tmpWriteBuf, tag, false);
+                    int curWrittenPages = writtenPagesCntr.incrementAndGet();
 
-                        updStores.computeIfAbsent(store, k -> new LongAdder()).increment();
-                    }
-                }
+                    snapshotMgr.onPageWrite(fullId, tmpWriteBuf, curWrittenPages, totalPagesToWrite);
 
-                doneFut.onDone((Void)null);
-            }
-            catch (Throwable e) {
-                doneFut.onDone(e);
+                    tmpWriteBuf.rewind();
+
+                    PageStore store = storeMgr.writeInternal(grpId, fullId.pageId(), tmpWriteBuf,
tag, false);
+
+                    updStores.computeIfAbsent(store, k -> new LongAdder()).increment();
+                }
             }
+
+            return pagesToRetry;
         }
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/a0fa79a8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
index 9861ef9..ea775dc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
@@ -1207,6 +1207,7 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl
imple
                 IgniteCacheDatabaseSharedManager dbMgr = ctx.database();
 
                 dbMgr.checkpointReadLock();
+
                 try {
                     Metas metas = getOrAllocatePartitionMetas();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/a0fa79a8/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 a518121..f8f3b57 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
@@ -177,6 +177,9 @@ public class PageMemoryImpl implements PageMemoryEx {
     /** Number of random pages that will be picked for eviction. */
     public static final int RANDOM_PAGES_EVICT_NUM = 5;
 
+    /** Try again tag. */
+    public static final int TRY_AGAIN_TAG = -1;
+
     /** Tracking io. */
     private static final TrackingPageIO trackingIO = TrackingPageIO.VERSIONS.latest();
 
@@ -377,9 +380,9 @@ public class PageMemoryImpl implements PageMemoryEx {
         if (throttlingPlc == ThrottlingPolicy.SPEED_BASED)
             writeThrottle = new PagesWriteSpeedBasedThrottle(this, cpProgressProvider, stateChecker,
log);
         else if (throttlingPlc == ThrottlingPolicy.TARGET_RATIO_BASED)
-            writeThrottle = new PagesWriteThrottle(this, cpProgressProvider, stateChecker,
false);
+            writeThrottle = new PagesWriteThrottle(this, cpProgressProvider, stateChecker,
false, log);
         else if (throttlingPlc == ThrottlingPolicy.CHECKPOINT_BUFFER_ONLY)
-            writeThrottle = new PagesWriteThrottle(this, null, stateChecker, true);
+            writeThrottle = new PagesWriteThrottle(this, null, stateChecker, true, log);
     }
 
     /** {@inheritDoc} */
@@ -1118,7 +1121,7 @@ public class PageMemoryImpl implements PageMemoryEx {
             }
         }
         else
-            return copyPageForCheckpoint(absPtr, fullId, outBuf, pageSingleAcquire, tracker)
? tag : null;
+            return copyPageForCheckpoint(absPtr, fullId, outBuf, pageSingleAcquire, tracker)
? tag : TRY_AGAIN_TAG;
     }
 
     /**
@@ -1128,6 +1131,8 @@ public class PageMemoryImpl implements PageMemoryEx {
      * @param pageSingleAcquire Page is acquired only once. We don't pin the page second
time (until page will not be
      * copied) in case checkpoint temporary buffer is used.
      * @param tracker Checkpoint statistics tracker.
+     *
+     * @return False if someone else holds lock on page.
      */
     private boolean copyPageForCheckpoint(
         long absPtr,
@@ -1139,7 +1144,10 @@ public class PageMemoryImpl implements PageMemoryEx {
         assert absPtr != 0;
         assert PageHeader.isAcquired(absPtr);
 
-        rwLock.writeLock(absPtr + PAGE_LOCK_OFFSET, OffheapReadWriteLock.TAG_LOCK_ALWAYS);
+        boolean locked = rwLock.tryWriteLock(absPtr + PAGE_LOCK_OFFSET, OffheapReadWriteLock.TAG_LOCK_ALWAYS);
+
+        if (!locked)
+            return false;
 
         try {
             long tmpRelPtr = PageHeader.tempBufferPointer(absPtr);
@@ -2346,6 +2354,8 @@ public class PageMemoryImpl implements PageMemoryEx {
 
             Integer tag = partGenerationMap.get(new GroupPartitionId(grpId, partId));
 
+            assert tag == null || tag >= 0 : "Negative tag=" + tag;
+
             return tag == null ? INIT_PART_GENERATION : tag;
         }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/a0fa79a8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteSpeedBasedThrottle.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteSpeedBasedThrottle.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteSpeedBasedThrottle.java
index 68fa529..2dd8127 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteSpeedBasedThrottle.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteSpeedBasedThrottle.java
@@ -24,6 +24,7 @@ import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.internal.processors.cache.persistence.CheckpointLockStateChecker;
 import org.apache.ignite.internal.processors.cache.persistence.CheckpointWriteProgressSupplier;
 import org.apache.ignite.internal.util.GridConcurrentHashSet;
+import org.apache.ignite.internal.util.typedef.internal.U;
 
 /**
  * Throttles threads that generate dirty pages during ongoing checkpoint.
@@ -113,10 +114,12 @@ public class PagesWriteSpeedBasedThrottle implements PagesWriteThrottlePolicy
{
      * @param stateChecker Checkpoint lock state provider.
      * @param log Logger.
      */
-    public PagesWriteSpeedBasedThrottle(PageMemoryImpl pageMemory,
-        CheckpointWriteProgressSupplier cpProgress,
-        CheckpointLockStateChecker stateChecker,
-        IgniteLogger log) {
+    public PagesWriteSpeedBasedThrottle(
+            PageMemoryImpl pageMemory,
+            CheckpointWriteProgressSupplier cpProgress,
+            CheckpointLockStateChecker stateChecker,
+            IgniteLogger log
+    ) {
         this.pageMemory = pageMemory;
         this.cpProgress = cpProgress;
         totalPages = pageMemory.totalPages();
@@ -229,6 +232,11 @@ public class PagesWriteSpeedBasedThrottle implements PagesWriteThrottlePolicy
{
      * @param throttleParkTimeNs the maximum number of nanoseconds to wait
      */
     protected void doPark(long throttleParkTimeNs) {
+        if (throttleParkTimeNs > LOGGING_THRESHOLD) {
+            U.warn(log, "Parking thread=" + Thread.currentThread().getName()
+                + " for timeout(ms)=" + (throttleParkTimeNs / 1_000_000));
+        }
+
         LockSupport.parkNanos(throttleParkTimeNs);
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/a0fa79a8/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
index 166cdcd..d5f4bd5 100644
--- 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
@@ -18,8 +18,10 @@ 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.IgniteLogger;
 import org.apache.ignite.internal.processors.cache.persistence.CheckpointLockStateChecker;
 import org.apache.ignite.internal.processors.cache.persistence.CheckpointWriteProgressSupplier;
+import org.apache.ignite.internal.util.typedef.internal.U;
 
 /**
  * Throttles threads that generate dirty pages during ongoing checkpoint.
@@ -50,21 +52,27 @@ public class PagesWriteThrottle implements PagesWriteThrottlePolicy {
     /** Counter for checkpoint buffer usage ratio throttling (we need a separate one due
to IGNITE-7751). */
     private final AtomicInteger inCheckpointBackoffCntr = new AtomicInteger(0);
 
+    /** Logger. */
+    private IgniteLogger log;
+
     /**
      * @param pageMemory Page memory.
      * @param cpProgress Database manager.
      * @param stateChecker checkpoint lock state checker.
      * @param throttleOnlyPagesInCheckpoint If true, throttle will only protect from checkpoint
buffer overflow.
+     * @param log Logger.
      */
     public PagesWriteThrottle(PageMemoryImpl pageMemory,
         CheckpointWriteProgressSupplier cpProgress,
         CheckpointLockStateChecker stateChecker,
-        boolean throttleOnlyPagesInCheckpoint
+        boolean throttleOnlyPagesInCheckpoint,
+        IgniteLogger log
     ) {
         this.pageMemory = pageMemory;
         this.cpProgress = cpProgress;
         this.stateChecker = stateChecker;
         this.throttleOnlyPagesInCheckpoint = throttleOnlyPagesInCheckpoint;
+        this.log = log;
 
         if (!throttleOnlyPagesInCheckpoint)
             assert cpProgress != null : "cpProgress must be not null if ratio based throttling
mode is used";
@@ -111,7 +119,14 @@ public class PagesWriteThrottle implements PagesWriteThrottlePolicy {
         if (shouldThrottle) {
             int throttleLevel = cntr.getAndIncrement();
 
-            LockSupport.parkNanos((long)(STARTING_THROTTLE_NANOS * Math.pow(BACKOFF_RATIO,
throttleLevel)));
+            long throttleParkTimeNs = (long) (STARTING_THROTTLE_NANOS * Math.pow(BACKOFF_RATIO,
throttleLevel));
+
+            if (throttleParkTimeNs > LOGGING_THRESHOLD) {
+                U.warn(log, "Parking thread=" + Thread.currentThread().getName()
+                    + " for timeout(ms)=" + (throttleParkTimeNs / 1_000_000));
+            }
+
+            LockSupport.parkNanos(throttleParkTimeNs);
         }
         else
             cntr.set(0);

http://git-wip-us.apache.org/repos/asf/ignite/blob/a0fa79a8/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottlePolicy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottlePolicy.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottlePolicy.java
index adeaa3d..53a8017 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottlePolicy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PagesWriteThrottlePolicy.java
@@ -17,10 +17,15 @@
 
 package org.apache.ignite.internal.processors.cache.persistence.pagemem;
 
+import java.util.concurrent.TimeUnit;
+
 /**
  * Throttling policy, encapsulates logic of delaying write operations.
  */
 public interface PagesWriteThrottlePolicy {
+    /** Max park time. */
+    public long LOGGING_THRESHOLD = TimeUnit.SECONDS.toNanos(10);
+
     /**
      * Callback to apply throttling delay.
      * @param isPageInCheckpoint flag indicating if current page is in scope of current checkpoint.

http://git-wip-us.apache.org/repos/asf/ignite/blob/a0fa79a8/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/CheckpointBufferDeadlockTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/CheckpointBufferDeadlockTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/CheckpointBufferDeadlockTest.java
new file mode 100644
index 0000000..2b5a65d
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/CheckpointBufferDeadlockTest.java
@@ -0,0 +1,358 @@
+/*
+* 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.db;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+import java.nio.file.OpenOption;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.LockSupport;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.failure.StopNodeFailureHandler;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.pagemem.FullPageId;
+import org.apache.ignite.internal.pagemem.PageIdAllocator;
+import org.apache.ignite.internal.pagemem.PageIdUtils;
+import org.apache.ignite.internal.pagemem.store.PageStore;
+import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+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.FilePageStoreManager;
+import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory;
+import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImpl;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+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;
+
+import static java.nio.file.StandardOpenOption.CREATE;
+import static java.nio.file.StandardOpenOption.READ;
+import static java.nio.file.StandardOpenOption.WRITE;
+
+/**
+ *
+ */
+public class CheckpointBufferDeadlockTest extends GridCommonAbstractTest {
+    /** Ip finder. */
+    private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+    /** Max size. */
+    private static final int MAX_SIZE = 500 * 1024 * 1024;
+
+    /** CP buffer size. */
+    private static final int CP_BUF_SIZE = 20 * 1024 * 1024;
+
+    /** Slow checkpoint enabled. */
+    private static final AtomicBoolean slowCheckpointEnabled = new AtomicBoolean(false);
+
+    /** Checkpoint park nanos. */
+    private static final int CHECKPOINT_PARK_NANOS = 50_000_000;
+
+    /** Entry byte chunk size. */
+    private static final int ENTRY_BYTE_CHUNK_SIZE = 900;
+
+    /** Pages touched under CP lock. */
+    private static final int PAGES_TOUCHED_UNDER_CP_LOCK = 20;
+
+    /** Slop load flag. */
+    private static final AtomicBoolean stop = new AtomicBoolean(false);
+
+    /** Checkpoint threads. */
+    private int checkpointThreads;
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws
Exception {
+        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+        cfg.setDiscoverySpi(new TcpDiscoverySpi().setIpFinder(IP_FINDER));
+
+        cfg.setDataStorageConfiguration(
+            new DataStorageConfiguration()
+                .setFileIOFactory(new SlowCheckpointFileIOFactory())
+                .setCheckpointThreads(checkpointThreads)
+                .setDefaultDataRegionConfiguration(
+                    new DataRegionConfiguration()
+                        .setPersistenceEnabled(true)
+                        .setMaxSize(MAX_SIZE)
+                        .setCheckpointPageBufferSize(CP_BUF_SIZE)
+                )
+        );
+
+        cfg.setFailureHandler(new StopNodeFailureHandler());
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        stop.set(false);
+
+        slowCheckpointEnabled.set(false);
+
+        stopAllGrids();
+
+        cleanPersistenceDir();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        stop.set(true);
+
+        slowCheckpointEnabled.set(false);
+
+        stopAllGrids();
+
+        cleanPersistenceDir();
+    }
+
+    /**
+     *
+     */
+    public void testFourCheckpointThreads() throws Exception {
+        checkpointThreads = 4;
+
+        runDeadlockScenario();
+    }
+
+    /**
+     *
+     */
+    public void testOneCheckpointThread() throws Exception {
+        checkpointThreads = 1;
+
+        runDeadlockScenario();
+    }
+
+    /**
+     *
+     */
+    private void runDeadlockScenario() throws Exception {
+        IgniteEx ig = startGrid(0);
+
+        ig.cluster().active(true);
+
+        GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)ig.context().cache().context().database();
+
+        FilePageStoreManager pageStoreMgr = (FilePageStoreManager)ig.context().cache().context().pageStore();
+
+        IgniteCache<Object, Object> singlePartCache = ig.getOrCreateCache(new CacheConfiguration<>()
+            .setName("single-part")
+            .setAffinity(new RendezvousAffinityFunction(false, 1)));
+
+        db.enableCheckpoints(false).get();
+
+        Thread.sleep(1_000);
+
+        try (IgniteDataStreamer<Object, Object> streamer = ig.dataStreamer(singlePartCache.getName()))
{
+            int entries = MAX_SIZE / ENTRY_BYTE_CHUNK_SIZE / 4;
+
+            for (int i = 0; i < entries; i++)
+                streamer.addData(i, new byte[ENTRY_BYTE_CHUNK_SIZE]);
+
+            streamer.flush();
+        }
+
+        slowCheckpointEnabled.set(true);
+        log.info(">>> Slow checkpoints enabled");
+
+        db.enableCheckpoints(true).get();
+
+        AtomicBoolean fail = new AtomicBoolean(false);
+
+        GridTestUtils.runMultiThreadedAsync(new Runnable() {
+            @Override public void run() {
+                int loops = 0;
+
+                while (!stop.get()) {
+                    if (loops % 10 == 0 && loops > 0 && loops < 500
|| loops % 500 == 0 && loops >= 500)
+                        log.info("Successfully completed " + loops + " loops");
+
+                    db.checkpointReadLock();
+
+                    try {
+                        Set<FullPageId> pickedPagesSet = new HashSet<>();
+
+                        PageStore store = pageStoreMgr.getStore(CU.cacheId("single-part"),
0);
+
+                        int pages = store.pages();
+
+                        DataRegion region = db.dataRegion(DataStorageConfiguration.DFLT_DATA_REG_DEFAULT_NAME);
+
+                        PageMemoryImpl pageMem = (PageMemoryImpl)region.pageMemory();
+
+                        while (pickedPagesSet.size() < PAGES_TOUCHED_UNDER_CP_LOCK) {
+                            int pageIdx = ThreadLocalRandom.current().nextInt(
+                                PAGES_TOUCHED_UNDER_CP_LOCK, pages - PAGES_TOUCHED_UNDER_CP_LOCK);
+
+                            long pageId = PageIdUtils.pageId(0, PageIdAllocator.FLAG_DATA,
pageIdx);
+
+                            pickedPagesSet.add(new FullPageId(pageId, CU.cacheId("single-part")));
+                        }
+
+                        List<FullPageId> pickedPages = new ArrayList<>(pickedPagesSet);
+
+                        assertEquals(PAGES_TOUCHED_UNDER_CP_LOCK, pickedPages.size());
+
+                        // Sort to avoid deadlocks on pages rw-locks.
+                        pickedPages.sort(new Comparator<FullPageId>() {
+                            @Override public int compare(FullPageId o1, FullPageId o2) {
+                                int cmp = Long.compare(o1.groupId(), o2.groupId());
+                                if (cmp != 0)
+                                    return cmp;
+
+                                return Long.compare(PageIdUtils.effectivePageId(o1.pageId()),
+                                    PageIdUtils.effectivePageId(o2.pageId()));
+                            }
+                        });
+
+                        List<Long> readLockedPages = new ArrayList<>();
+
+                        // Read lock many pages at once intentionally.
+                        for (int i = 0; i < PAGES_TOUCHED_UNDER_CP_LOCK / 2; i++) {
+                            FullPageId fpid = pickedPages.get(i);
+
+                            long page = pageMem.acquirePage(fpid.groupId(), fpid.pageId());
+
+                            long abs = pageMem.readLock(fpid.groupId(), fpid.pageId(), page);
+
+                            assertFalse(fpid.toString(), abs == 0);
+
+                            readLockedPages.add(page);
+                        }
+
+                        // Emulate writes to trigger throttling.
+                        for (int i = PAGES_TOUCHED_UNDER_CP_LOCK / 2; i < PAGES_TOUCHED_UNDER_CP_LOCK;
i++) {
+                            FullPageId fpid = pickedPages.get(i);
+
+                            long page = pageMem.acquirePage(fpid.groupId(), fpid.pageId());
+
+                            long abs = pageMem.writeLock(fpid.groupId(), fpid.pageId(), page);
+
+                            assertFalse(fpid.toString(), abs == 0);
+
+                            pageMem.writeUnlock(fpid.groupId(), fpid.pageId(), page, null,
true);
+
+                            pageMem.releasePage(fpid.groupId(), fpid.pageId(), page);
+                        }
+
+                        for (int i = 0; i < PAGES_TOUCHED_UNDER_CP_LOCK / 2; i++) {
+                            FullPageId fpid = pickedPages.get(i);
+
+                            pageMem.readUnlock(fpid.groupId(), fpid.pageId(), readLockedPages.get(i));
+
+                            pageMem.releasePage(fpid.groupId(), fpid.pageId(), readLockedPages.get(i));
+                        }
+                    }
+                    catch (Throwable e) {
+                        log.error("Error in loader thread", e);
+
+                        fail.set(true);
+                    }
+                    finally {
+                        db.checkpointReadUnlock();
+                    }
+
+                    loops++;
+                }
+
+            }
+        }, 10, "load-runner");
+
+        Thread.sleep(10_000); // Await for the start of throttling.
+
+        slowCheckpointEnabled.set(false);
+        log.info(">>> Slow checkpoints disabled");
+
+        assertFalse(fail.get());
+
+        forceCheckpoint(); // Previous checkpoint should eventually finish.
+    }
+
+    /**
+     * Create File I/O that emulates poor checkpoint write speed.
+     */
+    private static 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, CREATE, READ, WRITE);
+        }
+
+        /** {@inheritDoc} */
+        @Override public FileIO create(File file, OpenOption... openOption) throws IOException
{
+            final FileIO delegate = delegateFactory.create(file, openOption);
+
+            return new FileIODecorator(delegate) {
+                @Override public int write(ByteBuffer srcBuf) throws IOException {
+                    parkIfNeeded();
+
+                    return delegate.write(srcBuf);
+                }
+
+                @Override public int write(ByteBuffer srcBuf, long position) throws IOException
{
+                    parkIfNeeded();
+
+                    return delegate.write(srcBuf, position);
+                }
+
+                @Override public int write(byte[] buf, int off, int len) throws IOException
{
+                    parkIfNeeded();
+
+                    return delegate.write(buf, off, len);
+                }
+
+                /**
+                 * Parks current checkpoint thread if slow mode is enabled.
+                 */
+                private void parkIfNeeded() {
+                    if (slowCheckpointEnabled.get() && Thread.currentThread().getName().contains("checkpoint"))
+                        LockSupport.parkNanos(CHECKPOINT_PARK_NANOS);
+                }
+
+                /** {@inheritDoc} */
+                @Override public MappedByteBuffer map(int sizeBytes) throws IOException {
+                    return delegate.map(sizeBytes);
+                }
+            };
+        }
+    }
+
+
+}


Mime
View raw message