You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2024/09/13 13:54:43 UTC
(ignite-3) branch main updated: IGNITE-23115 Checkpoint single partition from a single thread (#4379)
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new c8f6929690 IGNITE-23115 Checkpoint single partition from a single thread (#4379)
c8f6929690 is described below
commit c8f692969004f38997fec60bff0f601e4cad948b
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Fri Sep 13 16:54:36 2024 +0300
IGNITE-23115 Checkpoint single partition from a single thread (#4379)
---
.../pagememory/persistence/GroupPartitionId.java | 31 +---
.../checkpoint/CheckpointDirtyPages.java | 100 +++-------
.../persistence/checkpoint/CheckpointManager.java | 13 +-
.../checkpoint/CheckpointPagesWriter.java | 205 ++++++++++++---------
.../checkpoint/CheckpointPagesWriterFactory.java | 7 +-
.../persistence/checkpoint/CheckpointWorkflow.java | 21 ++-
.../persistence/checkpoint/Checkpointer.java | 6 +-
.../checkpoint/DirtyPagesAndPartitions.java} | 32 ++--
.../checkpoint/CheckpointDirtyPagesTest.java | 127 +++++--------
.../checkpoint/CheckpointManagerTest.java | 17 +-
.../checkpoint/CheckpointPagesWriterTest.java | 57 ++++--
.../persistence/checkpoint/CheckpointTest.java | 7 +-
.../checkpoint/CheckpointWorkflowTest.java | 46 ++---
.../persistence/checkpoint/CheckpointerTest.java | 13 +-
...heckpointTest.java => TestCheckpointUtils.java} | 34 ++--
15 files changed, 332 insertions(+), 384 deletions(-)
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/GroupPartitionId.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/GroupPartitionId.java
index 73c97a3671..5c30c3c267 100644
--- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/GroupPartitionId.java
+++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/GroupPartitionId.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.pagememory.persistence;
+import org.apache.ignite.internal.pagememory.FullPageId;
import org.apache.ignite.internal.tostring.S;
/**
@@ -54,13 +55,11 @@ public class GroupPartitionId implements Comparable<GroupPartitionId> {
return partId;
}
- /** {@inheritDoc} */
@Override
public String toString() {
return S.toString(GroupPartitionId.class, this);
}
- /** {@inheritDoc} */
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -73,14 +72,10 @@ public class GroupPartitionId implements Comparable<GroupPartitionId> {
GroupPartitionId key = (GroupPartitionId) o;
- if (grpId != key.grpId) {
- return false;
- }
+ return grpId == key.grpId && partId == key.partId;
- return partId == key.partId;
}
- /** {@inheritDoc} */
@Override
public int hashCode() {
int result = grpId;
@@ -90,25 +85,19 @@ public class GroupPartitionId implements Comparable<GroupPartitionId> {
return result;
}
- /** {@inheritDoc} */
@Override
public int compareTo(GroupPartitionId o) {
- if (getGroupId() < o.getGroupId()) {
- return -1;
- }
+ int cmp = Integer.compare(getGroupId(), o.getGroupId());
- if (getGroupId() > o.getGroupId()) {
- return 1;
+ if (cmp != 0) {
+ return cmp;
}
- if (getPartitionId() < o.getPartitionId()) {
- return -1;
- }
-
- if (getPartitionId() > o.getPartitionId()) {
- return 1;
- }
+ return Integer.compare(getPartitionId(), o.getPartitionId());
+ }
- return 0;
+ /** Converts given full page ID to a {@link GroupPartitionId}. */
+ public static GroupPartitionId convert(FullPageId fullPageId) {
+ return new GroupPartitionId(fullPageId.groupId(), fullPageId.partitionId());
}
}
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.java
index b87a4188e4..86086ffa8f 100644
--- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.java
+++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.java
@@ -20,21 +20,18 @@ package org.apache.ignite.internal.pagememory.persistence.checkpoint;
import static java.util.Arrays.binarySearch;
import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.pagememory.util.PageIdUtils.pageId;
-import static org.apache.ignite.internal.pagememory.util.PageIdUtils.partitionId;
import java.util.Comparator;
import java.util.List;
import java.util.RandomAccess;
import org.apache.ignite.internal.lang.IgniteBiTuple;
import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
import org.apache.ignite.internal.util.IgniteConcurrentMultiPairQueue;
import org.jetbrains.annotations.Nullable;
-/**
- * Dirty pages of data regions, with sorted page IDs by {@link #DIRTY_PAGE_COMPARATOR} and unsorted partition IDs that should be
- * checkpointed.
- */
+/** Dirty pages of data regions, with sorted page IDs by {@link #DIRTY_PAGE_COMPARATOR} and partition IDs that should be checkpointed. */
class CheckpointDirtyPages {
/** Dirty page ID comparator by groupId -> partitionId -> pageIdx. */
static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
@@ -44,8 +41,8 @@ class CheckpointDirtyPages {
/** Empty checkpoint dirty pages. */
static final CheckpointDirtyPages EMPTY = new CheckpointDirtyPages(List.of());
- /** Dirty pages of data regions, with sorted page IDs by {@link #DIRTY_PAGE_COMPARATOR} and unsorted partition IDs. */
- private final List<DataRegionDirtyPages<FullPageId[]>> dirtyPages;
+ /** Dirty pages and partitions of data regions, with sorted dirty page IDs by {@link #DIRTY_PAGE_COMPARATOR}. */
+ private final List<DirtyPagesAndPartitions> dirtyPagesAndPartitions;
/** Total number of dirty page IDs. */
private final int dirtyPagesCount;
@@ -53,14 +50,15 @@ class CheckpointDirtyPages {
/**
* Constructor.
*
- * @param dirtyPages Dirty pages of data regions, with sorted page IDs by {@link #DIRTY_PAGE_COMPARATOR} and unsorted partition IDs.
+ * @param dirtyPagesAndPartitions Dirty pages and partitions of data regions, with sorted dirty page IDs by
+ * {@link #DIRTY_PAGE_COMPARATOR}. Expected list with {@link RandomAccess}.
*/
- public CheckpointDirtyPages(List<DataRegionDirtyPages<FullPageId[]>> dirtyPages) {
- assert dirtyPages instanceof RandomAccess : dirtyPages;
+ CheckpointDirtyPages(List<DirtyPagesAndPartitions> dirtyPagesAndPartitions) {
+ assert dirtyPagesAndPartitions instanceof RandomAccess : dirtyPagesAndPartitions;
- this.dirtyPages = dirtyPages;
+ this.dirtyPagesAndPartitions = dirtyPagesAndPartitions;
- dirtyPagesCount = dirtyPages.stream().mapToInt(pages -> pages.dirtyPages.length).sum();
+ dirtyPagesCount = dirtyPagesAndPartitions.stream().mapToInt(pages -> pages.dirtyPages.length).sum();
}
/**
@@ -70,15 +68,16 @@ class CheckpointDirtyPages {
return dirtyPagesCount;
}
- /**
- * Returns a queue of dirty page IDs to be written to a checkpoint.
- */
- public IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> toDirtyPageIdQueue() {
- List<IgniteBiTuple<PersistentPageMemory, FullPageId[]>> dirtyPageIds = dirtyPages.stream()
- .map(pages -> new IgniteBiTuple<>(pages.pageMemory, pages.dirtyPages))
+ /** Creates a concurrent queue of dirty partitions to be written to at checkpoint. */
+ public IgniteConcurrentMultiPairQueue<PersistentPageMemory, GroupPartitionId> toDirtyPartitionQueue() {
+ List<IgniteBiTuple<PersistentPageMemory, GroupPartitionId[]>> dirtyPartitions = dirtyPagesAndPartitions.stream()
+ .map(dirtyPagesAndPartitions -> new IgniteBiTuple<>(
+ dirtyPagesAndPartitions.pageMemory,
+ dirtyPagesAndPartitions.dirtyPartitions.toArray(GroupPartitionId[]::new))
+ )
.collect(toList());
- return new IgniteConcurrentMultiPairQueue<>(dirtyPageIds);
+ return new IgniteConcurrentMultiPairQueue<>(dirtyPartitions);
}
/**
@@ -89,8 +88,8 @@ class CheckpointDirtyPages {
* @param partId Partition ID.
*/
public @Nullable CheckpointDirtyPagesView getPartitionView(PersistentPageMemory pageMemory, int grpId, int partId) {
- for (int i = 0; i < dirtyPages.size(); i++) {
- if (dirtyPages.get(i).pageMemory == pageMemory) {
+ for (int i = 0; i < dirtyPagesAndPartitions.size(); i++) {
+ if (dirtyPagesAndPartitions.get(i).pageMemory == pageMemory) {
return getPartitionView(i, grpId, partId);
}
}
@@ -102,7 +101,7 @@ class CheckpointDirtyPages {
FullPageId startPageId = new FullPageId(pageId(partId, (byte) 0, 0), grpId);
FullPageId endPageId = new FullPageId(pageId(partId + 1, (byte) 0, 0), grpId);
- FullPageId[] pageIds = dirtyPages.get(dirtyPagesIdx).dirtyPages;
+ FullPageId[] pageIds = dirtyPagesAndPartitions.get(dirtyPagesIdx).dirtyPages;
int fromIndex = binarySearch(pageIds, startPageId, DIRTY_PAGE_COMPARATOR);
@@ -119,51 +118,12 @@ class CheckpointDirtyPages {
return new CheckpointDirtyPagesView(dirtyPagesIdx, fromIndex, toIndex);
}
- /**
- * Looks for the next dirty page IDs view from the current one, {@code null} if not found.
- *
- * @param currentView Current view to dirty pages, {@code null} to get first.
- */
- public @Nullable CheckpointDirtyPagesView nextPartitionView(@Nullable CheckpointDirtyPagesView currentView) {
- assert currentView == null || currentView.owner() == this : currentView;
-
- if (dirtyPages.isEmpty()) {
- return null;
- }
-
- int regionIndex;
- int fromPosition;
-
- if (currentView == null) {
- regionIndex = 0;
- fromPosition = 0;
- } else {
- regionIndex = currentView.needsNextRegion() ? currentView.regionIndex + 1 : currentView.regionIndex;
- fromPosition = currentView.needsNextRegion() ? 0 : currentView.toPosition;
- }
-
- if (regionIndex >= dirtyPages.size()) {
- return null;
- }
-
- FullPageId[] pageIds = dirtyPages.get(regionIndex).dirtyPages;
-
- FullPageId startPageId = pageIds[fromPosition];
- FullPageId endPageId = new FullPageId(pageId(partitionId(startPageId.pageId()) + 1, (byte) 0, 0), startPageId.groupId());
-
- int toPosition = binarySearch(pageIds, fromPosition, pageIds.length, endPageId, DIRTY_PAGE_COMPARATOR);
-
- toPosition = toPosition > 0 ? toPosition : -toPosition - 1;
-
- return new CheckpointDirtyPagesView(regionIndex, fromPosition, toPosition);
- }
-
/**
* Returns a full list of {@link PersistentPageMemory} instances, for which there exist at least a single dirty partition in current
* checkpoint.
*/
List<PersistentPageMemory> dirtyPageMemoryInstances() {
- return this.dirtyPages.stream().map(p -> p.pageMemory).collect(toList());
+ return this.dirtyPagesAndPartitions.stream().map(p -> p.pageMemory).collect(toList());
}
/**
@@ -173,7 +133,7 @@ class CheckpointDirtyPages {
* <p>Thread safe.
*/
class CheckpointDirtyPagesView {
- /** Element index in {@link CheckpointDirtyPages#dirtyPages}. */
+ /** Element index in {@link CheckpointDirtyPages#dirtyPagesAndPartitions}. */
private final int regionIndex;
/** Starting position (inclusive) of the dirty page within the element at {@link #regionIndex}. */
@@ -185,7 +145,7 @@ class CheckpointDirtyPages {
/**
* Private constructor.
*
- * @param regionIndex Element index in {@link CheckpointDirtyPages#dirtyPages}.
+ * @param regionIndex Element index in {@link CheckpointDirtyPages#dirtyPagesAndPartitions}.
* @param fromPosition Starting position (inclusive) of the dirty page within the element at {@link #regionIndex}.
* @param toPosition End position (exclusive) of the dirty page within the element at {@link #regionIndex}.
*/
@@ -201,14 +161,14 @@ class CheckpointDirtyPages {
* @param index Dirty page index.
*/
public FullPageId get(int index) {
- return dirtyPages.get(this.regionIndex).dirtyPages[fromPosition + index];
+ return dirtyPagesAndPartitions.get(this.regionIndex).dirtyPages[fromPosition + index];
}
/**
* Returns the page memory for view.
*/
public PersistentPageMemory pageMemory() {
- return dirtyPages.get(regionIndex).pageMemory;
+ return dirtyPagesAndPartitions.get(regionIndex).pageMemory;
}
/**
@@ -217,14 +177,6 @@ class CheckpointDirtyPages {
public int size() {
return toPosition - fromPosition;
}
-
- private CheckpointDirtyPages owner() {
- return CheckpointDirtyPages.this;
- }
-
- private boolean needsNextRegion() {
- return toPosition == dirtyPages.get(regionIndex).dirtyPages.length;
- }
}
private static boolean equalsByGroupAndPartition(FullPageId pageId0, FullPageId pageId1) {
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointManager.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointManager.java
index 4e9c38f46a..4d85b72087 100644
--- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointManager.java
+++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointManager.java
@@ -317,7 +317,18 @@ public class CheckpointManager {
CompletableFuture<DeltaFilePageStoreIo> deltaFilePageStoreFuture = filePageStore.getOrCreateNewDeltaFile(
index -> filePageStoreManager.tmpDeltaFilePageStorePath(pageId.groupId(), pageId.partitionId(), index),
- () -> pageIndexesForDeltaFilePageStore(pagesToWrite.getPartitionView(pageMemory, pageId.groupId(), pageId.partitionId()))
+ () -> {
+ CheckpointDirtyPagesView partitionView = pagesToWrite.getPartitionView(
+ pageMemory,
+ pageId.groupId(),
+ pageId.partitionId()
+ );
+
+ assert partitionView != null : String.format("Unable to find view for dirty pages: [patitionId=%s, pageMemory=%s]",
+ GroupPartitionId.convert(pageId), pageMemory);
+
+ return pageIndexesForDeltaFilePageStore(partitionView);
+ }
);
deltaFilePageStoreFuture.join().write(pageId.pageId(), pageBuf, calculateCrc);
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriter.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriter.java
index e1ae8085e7..bf93a24494 100644
--- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriter.java
+++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriter.java
@@ -23,7 +23,6 @@ import static org.apache.ignite.internal.pagememory.io.PageIo.getVersion;
import static org.apache.ignite.internal.pagememory.persistence.PartitionMeta.partitionMetaPageId;
import static org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory.TRY_AGAIN_TAG;
import static org.apache.ignite.internal.pagememory.util.PageIdUtils.flag;
-import static org.apache.ignite.internal.util.IgniteConcurrentMultiPairQueue.EMPTY;
import static org.apache.ignite.internal.util.StringUtils.hexLong;
import java.nio.ByteBuffer;
@@ -31,6 +30,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.LongAdder;
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.pagememory.persistence.PartitionMeta.Partition
import org.apache.ignite.internal.pagememory.persistence.PartitionMetaManager;
import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
import org.apache.ignite.internal.pagememory.persistence.WriteDirtyPage;
+import org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.CheckpointDirtyPagesView;
import org.apache.ignite.internal.pagememory.persistence.io.PartitionMetaIo;
import org.apache.ignite.internal.util.IgniteConcurrentMultiPairQueue;
import org.apache.ignite.internal.util.IgniteConcurrentMultiPairQueue.Result;
@@ -55,6 +56,7 @@ import org.jetbrains.annotations.Nullable;
/**
* Implementation of page writer which able to store pages to disk during checkpoint.
*/
+// TODO: IGNITE-23203 Write retry pages in multiple threads
public class CheckpointPagesWriter implements Runnable {
/** Logger. */
private static final IgniteLogger LOG = Loggers.forClass(CheckpointPagesWriter.class);
@@ -63,25 +65,20 @@ public class CheckpointPagesWriter implements Runnable {
* Size of a batch of pages that we drain from a single checkpoint buffer at the same time. The value of {@code 10} is chosen
* arbitrarily. We may reconsider it in <a href="https://issues.apache.org/jira/browse/IGNITE-23106">IGNITE-23106</a> if necessary.
*
- * @see #drainCheckpointBuffers(ByteBuffer, Map)
+ * @see #drainCheckpointBuffers(ByteBuffer)
*/
private static final int CP_BUFFER_PAGES_BATCH_THRESHOLD = 10;
/** Checkpoint specific metrics tracker. */
private final CheckpointMetricsTracker tracker;
- /**
- * Queue of dirty page IDs to write under this task.
- *
- * <p>Overall pages to write may be greater than this queue, since it may be necessary to retire write some pages due to unsuccessful
- * page write lock acquisition
- */
- private final IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> writePageIds;
+ /** Queue of dirty partitions IDs to write under this task. */
+ private final IgniteConcurrentMultiPairQueue<PersistentPageMemory, GroupPartitionId> dirtyPartitionQueue;
/**
* List of {@link PersistentPageMemory} instances that have dirty partitions in current checkpoint.
*
- * @see #drainCheckpointBuffers(ByteBuffer, Map)
+ * @see #drainCheckpointBuffers(ByteBuffer)
*/
private final List<PersistentPageMemory> pageMemoryList;
@@ -116,7 +113,7 @@ public class CheckpointPagesWriter implements Runnable {
* Creates task for write pages.
*
* @param tracker Checkpoint metrics tracker.
- * @param writePageIds Queue of dirty page IDs to write.
+ * @param dirtyPartitionQueue Queue of dirty partition IDs to write.
* @param pageMemoryList List of {@link PersistentPageMemory} instances that have dirty partitions in current checkpoint.
* @param updatedPartitions Updated partitions.
* @param doneFut Done future.
@@ -130,7 +127,7 @@ public class CheckpointPagesWriter implements Runnable {
*/
CheckpointPagesWriter(
CheckpointMetricsTracker tracker,
- IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> writePageIds,
+ IgniteConcurrentMultiPairQueue<PersistentPageMemory, GroupPartitionId> dirtyPartitionQueue,
List<PersistentPageMemory> pageMemoryList,
ConcurrentMap<GroupPartitionId, LongAdder> updatedPartitions,
CompletableFuture<?> doneFut,
@@ -143,7 +140,7 @@ public class CheckpointPagesWriter implements Runnable {
BooleanSupplier shutdownNow
) {
this.tracker = tracker;
- this.writePageIds = writePageIds;
+ this.dirtyPartitionQueue = dirtyPartitionQueue;
this.pageMemoryList = pageMemoryList;
this.updatedPartitions = updatedPartitions;
this.doneFut = doneFut;
@@ -159,15 +156,26 @@ public class CheckpointPagesWriter implements Runnable {
@Override
public void run() {
try {
- IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> pageIdsToRetry = writePages(writePageIds);
+ Map<PersistentPageMemory, List<FullPageId>> pageIdsToRetry = new HashMap<>();
- while (!pageIdsToRetry.isEmpty()) {
- if (LOG.isInfoEnabled()) {
- LOG.info("Checkpoint pages were not written yet due to "
- + "unsuccessful page write lock acquisition and will be retried [pageCount={}]", pageIdsToRetry.size());
- }
+ ByteBuffer tmpWriteBuf = threadBuf.get();
+
+ var queueResult = new Result<PersistentPageMemory, GroupPartitionId>();
+
+ while (!shutdownNow.getAsBoolean() && dirtyPartitionQueue.next(queueResult)) {
+ updateHeartbeat.run();
+
+ PersistentPageMemory pageMemory = queueResult.getKey();
+
+ PageStoreWriter pageStoreWriter = createPageStoreWriter(pageMemory, pageIdsToRetry);
+
+ writeDirtyPages(pageMemory, queueResult.getValue(), tmpWriteBuf, pageStoreWriter);
+ }
- pageIdsToRetry = writePages(pageIdsToRetry);
+ while (!shutdownNow.getAsBoolean() && !pageIdsToRetry.isEmpty()) {
+ updateHeartbeat.run();
+
+ pageIdsToRetry = writeRetryDirtyPages(pageIdsToRetry, tmpWriteBuf);
}
doneFut.complete(null);
@@ -176,90 +184,105 @@ public class CheckpointPagesWriter implements Runnable {
}
}
- /**
- * Writes dirty pages.
- *
- * @param writePageIds Queue of dirty page IDs to write.
- * @return pagesToRetry Queue dirty page IDs which should be retried.
- */
- private IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> writePages(
- IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> writePageIds
+ private void writeDirtyPages(
+ PersistentPageMemory pageMemory,
+ GroupPartitionId partitionId,
+ ByteBuffer tmpWriteBuf,
+ PageStoreWriter pageStoreWriter
) throws IgniteInternalCheckedException {
- Map<PersistentPageMemory, List<FullPageId>> pageIdsToRetry = new HashMap<>();
+ CheckpointDirtyPagesView checkpointDirtyPagesView = checkpointDirtyPagesView(pageMemory, partitionId);
- Map<PersistentPageMemory, PageStoreWriter> pageStoreWriters = new HashMap<>();
+ checkpointProgress.onStartPartitionProcessing(partitionId);
- // Page store writers for checkpoint buffer pages are located in a separate map, because they would not add elements to a
- // "pageIdsToRetry" map. Instead, they would ignore unsuccessful write lock acquisitions. It's implemented this way in order to
- // avoid duplicates in "pageIdsToRetry", there must only be a single source of pages to achieve that.
- Map<PersistentPageMemory, PageStoreWriter> cpBufferPageStoreWriters = new HashMap<>();
+ try {
+ if (shouldWriteMetaPage(partitionId)) {
+ writePartitionMeta(pageMemory, partitionId, tmpWriteBuf.rewind());
+ }
- ByteBuffer tmpWriteBuf = threadBuf.get();
+ for (int i = 0; i < checkpointDirtyPagesView.size() && !shutdownNow.getAsBoolean(); i++) {
+ updateHeartbeat.run();
- Result<PersistentPageMemory, FullPageId> queueResult = new Result<>();
+ FullPageId pageId = checkpointDirtyPagesView.get(i);
- GroupPartitionId partitionId = null;
+ if (pageId.pageIdx() == 0) {
+ // Skip meta-pages, they are written by "writePartitionMeta".
+ continue;
+ }
- try {
- // TODO https://issues.apache.org/jira/browse/IGNITE-23115 Try to write file per thread.
- while (!shutdownNow.getAsBoolean() && writePageIds.next(queueResult)) {
- updateHeartbeat.run();
+ writeDirtyPage(pageMemory, pageId, tmpWriteBuf, pageStoreWriter);
+ }
+ } finally {
+ checkpointProgress.onFinishPartitionProcessing(partitionId);
+ }
+ }
- FullPageId fullId = queueResult.getValue();
+ private void writeDirtyPage(
+ PersistentPageMemory pageMemory,
+ FullPageId pageId,
+ ByteBuffer tmpWriteBuf,
+ PageStoreWriter pageStoreWriter
+ ) throws IgniteInternalCheckedException {
+ // Should also be done for partitions that will be destroyed to remove their pages from the data region.
+ pageMemory.checkpointWritePage(pageId, tmpWriteBuf.rewind(), pageStoreWriter, tracker);
- PersistentPageMemory pageMemory = queueResult.getKey();
+ drainCheckpointBuffers(tmpWriteBuf);
+ }
- if (hasPartitionChanged(partitionId, fullId)) {
- GroupPartitionId newPartitionId = toPartitionId(fullId);
+ private Map<PersistentPageMemory, List<FullPageId>> writeRetryDirtyPages(
+ Map<PersistentPageMemory, List<FullPageId>> pageIdsToRetry,
+ ByteBuffer tmpWriteBuf
+ ) throws IgniteInternalCheckedException {
+ if (LOG.isInfoEnabled()) {
+ int pageCount = pageIdsToRetry.values().stream().mapToInt(List::size).sum();
- // Starting for the new partition.
- checkpointProgress.onStartPartitionProcessing(newPartitionId);
+ LOG.info("Checkpoint pages were not written yet due to "
+ + "unsuccessful page write lock acquisition and will be retried [pageCount={}]", pageCount);
+ }
- if (partitionId != null) {
- // Finishing for the previous partition.
- // TODO https://issues.apache.org/jira/browse/IGNITE-23105 Reimplement partition destruction awaiting.
- checkpointProgress.onFinishPartitionProcessing(partitionId);
- }
+ var newPageIdsToRetry = new HashMap<PersistentPageMemory, List<FullPageId>>();
- partitionId = newPartitionId;
+ for (Entry<PersistentPageMemory, List<FullPageId>> entry : pageIdsToRetry.entrySet()) {
+ PersistentPageMemory pageMemory = entry.getKey();
- if (shouldWriteMetaPage(partitionId)) {
- writePartitionMeta(pageMemory, partitionId, tmpWriteBuf.rewind());
+ PageStoreWriter pageStoreWriter = createPageStoreWriter(pageMemory, newPageIdsToRetry);
+
+ GroupPartitionId partitionId = null;
+
+ try {
+ for (FullPageId pageId : entry.getValue()) {
+ if (shutdownNow.getAsBoolean()) {
+ return Map.of();
}
- }
- PageStoreWriter pageStoreWriter = pageStoreWriters.computeIfAbsent(
- pageMemory,
- pm -> createPageStoreWriter(pm, pageIdsToRetry)
- );
+ updateHeartbeat.run();
+
+ if (partitionIdChanged(partitionId, pageId)) {
+ if (partitionId != null) {
+ checkpointProgress.onFinishPartitionProcessing(partitionId);
+ }
- if (fullId.pageIdx() == 0) {
- // Skip meta-pages, they are written by "writePartitionMeta".
- continue;
- }
+ partitionId = GroupPartitionId.convert(pageId);
- // Should also be done for partitions that will be destroyed to remove their pages from the data region.
- pageMemory.checkpointWritePage(fullId, tmpWriteBuf.rewind(), pageStoreWriter, tracker);
+ checkpointProgress.onStartPartitionProcessing(partitionId);
+ }
- drainCheckpointBuffers(tmpWriteBuf, cpBufferPageStoreWriters);
- }
- } finally {
- if (partitionId != null) {
- checkpointProgress.onFinishPartitionProcessing(partitionId);
+ writeDirtyPage(pageMemory, pageId, tmpWriteBuf, pageStoreWriter);
+ }
+ } finally {
+ if (partitionId != null) {
+ checkpointProgress.onFinishPartitionProcessing(partitionId);
+ }
}
}
- return pageIdsToRetry.isEmpty() ? EMPTY : new IgniteConcurrentMultiPairQueue<>(pageIdsToRetry);
+ return newPageIdsToRetry;
}
/**
* Checkpoints parts of checkpoint buffers if they are close to overflow. Uses
* {@link PersistentPageMemory#isCpBufferOverflowThresholdExceeded()} to detect that.
*/
- private void drainCheckpointBuffers(
- ByteBuffer tmpWriteBuf,
- Map<PersistentPageMemory, PageStoreWriter> pageStoreWriters
- ) throws IgniteInternalCheckedException {
+ private void drainCheckpointBuffers(ByteBuffer tmpWriteBuf) throws IgniteInternalCheckedException {
PageStoreWriter pageStoreWriter;
boolean retry = true;
@@ -288,16 +311,13 @@ public class CheckpointPagesWriter implements Runnable {
break;
}
- GroupPartitionId partitionId = toPartitionId(cpPageId);
+ GroupPartitionId partitionId = GroupPartitionId.convert(cpPageId);
if (shouldWriteMetaPage(partitionId)) {
writePartitionMeta(pageMemory, partitionId, tmpWriteBuf.rewind());
}
- pageStoreWriter = pageStoreWriters.computeIfAbsent(
- pageMemory,
- pm -> createPageStoreWriter(pm, null)
- );
+ pageStoreWriter = createPageStoreWriter(pageMemory, null);
pageMemory.checkpointWritePage(cpPageId, tmpWriteBuf.rewind(), pageStoreWriter, tracker);
}
@@ -350,7 +370,7 @@ public class CheckpointPagesWriter implements Runnable {
pageWriter.write(pageMemory, fullPageId, buf);
- updatedPartitions.get(toPartitionId(fullPageId)).increment();
+ updatedPartitions.get(GroupPartitionId.convert(fullPageId)).increment();
};
}
@@ -381,11 +401,24 @@ public class CheckpointPagesWriter implements Runnable {
updateHeartbeat.run();
}
- private static boolean hasPartitionChanged(@Nullable GroupPartitionId partitionId, FullPageId pageId) {
- return partitionId == null || partitionId.getPartitionId() != pageId.partitionId() || partitionId.getGroupId() != pageId.groupId();
+ private CheckpointDirtyPagesView checkpointDirtyPagesView(PersistentPageMemory pageMemory, GroupPartitionId partitionId) {
+ CheckpointDirtyPages checkpointDirtyPages = checkpointProgress.pagesToWrite();
+
+ assert checkpointDirtyPages != null;
+
+ CheckpointDirtyPagesView partitionView = checkpointDirtyPages.getPartitionView(
+ pageMemory,
+ partitionId.getGroupId(),
+ partitionId.getPartitionId()
+ );
+
+ assert partitionView != null : String.format("Unable to find view for dirty pages: [patitionId=%s, pageMemory=%s]", partitionId,
+ pageMemory);
+
+ return partitionView;
}
- private static GroupPartitionId toPartitionId(FullPageId pageId) {
- return new GroupPartitionId(pageId.groupId(), pageId.partitionId());
+ private static boolean partitionIdChanged(@Nullable GroupPartitionId partitionId, FullPageId pageId) {
+ return partitionId == null || partitionId.getGroupId() != pageId.groupId() || partitionId.getPartitionId() != pageId.partitionId();
}
}
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriterFactory.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriterFactory.java
index 301313ccd7..c0c05e5ca1 100644
--- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriterFactory.java
+++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriterFactory.java
@@ -24,7 +24,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.BooleanSupplier;
-import org.apache.ignite.internal.pagememory.FullPageId;
import org.apache.ignite.internal.pagememory.io.PageIoRegistry;
import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
import org.apache.ignite.internal.pagememory.persistence.PartitionMetaManager;
@@ -82,7 +81,7 @@ public class CheckpointPagesWriterFactory {
* Returns instance of page checkpoint writer.
*
* @param tracker Checkpoint metrics tracker.
- * @param dirtyPageIdQueue Checkpoint dirty page ID queue to write.
+ * @param dirtyPartitionQueue Checkpoint dirty partition ID queue to write.
* @param pageMemoryList List of {@link PersistentPageMemory} instances that have dirty partitions in current checkpoint.
* @param updatedPartitions Updated partitions.
* @param doneWriteFut Write done future.
@@ -92,7 +91,7 @@ public class CheckpointPagesWriterFactory {
*/
CheckpointPagesWriter build(
CheckpointMetricsTracker tracker,
- IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> dirtyPageIdQueue,
+ IgniteConcurrentMultiPairQueue<PersistentPageMemory, GroupPartitionId> dirtyPartitionQueue,
List<PersistentPageMemory> pageMemoryList,
ConcurrentMap<GroupPartitionId, LongAdder> updatedPartitions,
CompletableFuture<?> doneWriteFut,
@@ -103,7 +102,7 @@ public class CheckpointPagesWriterFactory {
) {
return new CheckpointPagesWriter(
tracker,
- dirtyPageIdQueue,
+ dirtyPartitionQueue,
pageMemoryList,
updatedPartitions,
doneWriteFut,
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointWorkflow.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointWorkflow.java
index bc074fc851..ce5a4c9dbc 100644
--- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointWorkflow.java
+++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointWorkflow.java
@@ -37,6 +37,7 @@ import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermin
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -57,6 +58,7 @@ import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.pagememory.DataRegion;
import org.apache.ignite.internal.pagememory.FullPageId;
import org.apache.ignite.internal.pagememory.PageMemory;
+import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.internal.util.CollectionUtils;
@@ -409,13 +411,15 @@ class CheckpointWorkflow {
CheckpointDirtyPages createAndSortCheckpointDirtyPages(
DataRegionsDirtyPages dataRegionsDirtyPages
) throws IgniteInternalCheckedException {
- List<DataRegionDirtyPages<FullPageId[]>> checkpointDirtyPages = new ArrayList<>();
+ var checkpointDirtyPages = new ArrayList<DirtyPagesAndPartitions>();
int realPagesArrSize = 0;
- // Collect arrays of dirty pages for sorting.
+ // Collects dirty pages into an array (then we will sort them) and collects dirty partitions.
for (DataRegionDirtyPages<Collection<FullPageId>> dataRegionDirtyPages : dataRegionsDirtyPages.dirtyPages) {
- FullPageId[] pageIds = new FullPageId[dataRegionDirtyPages.dirtyPages.size()];
+ var pageIds = new FullPageId[dataRegionDirtyPages.dirtyPages.size()];
+
+ var partitionIds = new HashSet<GroupPartitionId>();
int pagePos = 0;
@@ -424,6 +428,7 @@ class CheckpointWorkflow {
"Incorrect estimated dirty pages number: " + dataRegionsDirtyPages.dirtyPageCount;
pageIds[pagePos++] = dirtyPage;
+ partitionIds.add(GroupPartitionId.convert(dirtyPage));
}
// Some pages may have been already replaced.
@@ -433,20 +438,20 @@ class CheckpointWorkflow {
pageIds = Arrays.copyOf(pageIds, pagePos);
}
- checkpointDirtyPages.add(new DataRegionDirtyPages<>(dataRegionDirtyPages.pageMemory, pageIds));
+ checkpointDirtyPages.add(new DirtyPagesAndPartitions(dataRegionDirtyPages.pageMemory, pageIds, partitionIds));
}
// Add tasks to sort arrays of dirty page IDs in parallel if their number is greater than or equal to PARALLEL_SORT_THRESHOLD.
List<ForkJoinTask<?>> parallelSortTasks = checkpointDirtyPages.stream()
- .map(dataRegionDirtyPages -> dataRegionDirtyPages.dirtyPages)
+ .map(dirtyPagesAndPartitions -> dirtyPagesAndPartitions.dirtyPages)
.filter(pageIds -> pageIds.length >= PARALLEL_SORT_THRESHOLD)
.map(pageIds -> parallelSortThreadPool.submit(() -> Arrays.parallelSort(pageIds, DIRTY_PAGE_COMPARATOR)))
.collect(toList());
// Sort arrays of dirty page IDs if their number is less than PARALLEL_SORT_THRESHOLD.
- for (DataRegionDirtyPages<FullPageId[]> dataRegionDirtyPages : checkpointDirtyPages) {
- if (dataRegionDirtyPages.dirtyPages.length < PARALLEL_SORT_THRESHOLD) {
- Arrays.sort(dataRegionDirtyPages.dirtyPages, DIRTY_PAGE_COMPARATOR);
+ for (DirtyPagesAndPartitions dirtyPagesAndPartitions : checkpointDirtyPages) {
+ if (dirtyPagesAndPartitions.dirtyPages.length < PARALLEL_SORT_THRESHOLD) {
+ Arrays.sort(dirtyPagesAndPartitions.dirtyPages, DIRTY_PAGE_COMPARATOR);
}
}
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/Checkpointer.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/Checkpointer.java
index 39a7bbfe69..4b8c5457d8 100644
--- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/Checkpointer.java
+++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/Checkpointer.java
@@ -57,7 +57,6 @@ import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.pagememory.DataRegion;
-import org.apache.ignite.internal.pagememory.FullPageId;
import org.apache.ignite.internal.pagememory.configuration.schema.PageMemoryCheckpointConfiguration;
import org.apache.ignite.internal.pagememory.configuration.schema.PageMemoryCheckpointView;
import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
@@ -458,12 +457,13 @@ public class Checkpointer extends IgniteWorker {
List<PersistentPageMemory> pageMemoryList = checkpointDirtyPages.dirtyPageMemoryInstances();
- IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> writePageIds = checkpointDirtyPages.toDirtyPageIdQueue();
+ IgniteConcurrentMultiPairQueue<PersistentPageMemory, GroupPartitionId> dirtyPartitionQueue
+ = checkpointDirtyPages.toDirtyPartitionQueue();
for (int i = 0; i < checkpointWritePageThreads; i++) {
CheckpointPagesWriter write = checkpointPagesWriterFactory.build(
tracker,
- writePageIds,
+ dirtyPartitionQueue,
pageMemoryList,
updatedPartitions,
futures[i] = new CompletableFuture<>(),
diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointTest.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/DirtyPagesAndPartitions.java
similarity index 51%
copy from modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointTest.java
copy to modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/DirtyPagesAndPartitions.java
index 6f5f6c3fc5..eadbbd2802 100644
--- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointTest.java
+++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/DirtyPagesAndPartitions.java
@@ -17,32 +17,22 @@
package org.apache.ignite.internal.pagememory.persistence.checkpoint;
-import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.EMPTY;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.mock;
-
-import java.util.List;
+import java.util.Set;
import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
-import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
-import org.junit.jupiter.api.Test;
-/**
- * For {@link Checkpoint} testing.
- */
-public class CheckpointTest extends BaseIgniteAbstractTest {
- @Test
- void testHasDelta() {
- CheckpointProgressImpl progress = mock(CheckpointProgressImpl.class);
+/** Container of dirty pages and partitions. */
+class DirtyPagesAndPartitions {
+ final PersistentPageMemory pageMemory;
- assertFalse(new Checkpoint(EMPTY, progress).hasDelta());
+ final FullPageId[] dirtyPages;
- DataRegionDirtyPages<FullPageId[]> biTuple = new DataRegionDirtyPages<>(
- mock(PersistentPageMemory.class),
- new FullPageId[]{new FullPageId(0, 1)}
- );
+ final Set<GroupPartitionId> dirtyPartitions;
- assertTrue(new Checkpoint(new CheckpointDirtyPages(List.of(biTuple)), progress).hasDelta());
+ DirtyPagesAndPartitions(PersistentPageMemory pageMemory, FullPageId[] dirtyPages, Set<GroupPartitionId> dirtyPartitions) {
+ this.pageMemory = pageMemory;
+ this.dirtyPages = dirtyPages;
+ this.dirtyPartitions = dirtyPartitions;
}
}
diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPagesTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPagesTest.java
index 415d97fc71..05c1647897 100644
--- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPagesTest.java
+++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPagesTest.java
@@ -30,13 +30,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.function.Predicate;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import org.apache.ignite.internal.lang.IgniteBiTuple;
import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
import org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.CheckpointDirtyPagesView;
import org.apache.ignite.internal.pagememory.util.PageIdUtils;
@@ -51,8 +51,8 @@ import org.junit.jupiter.api.Test;
public class CheckpointDirtyPagesTest extends BaseIgniteAbstractTest {
@Test
void testDirtyPagesCount() {
- DataRegionDirtyPages<FullPageId[]> dirtyPages0 = createDirtyPages(of(0, 0, 0), of(0, 0, 1));
- DataRegionDirtyPages<FullPageId[]> dirtyPages1 = createDirtyPages(of(1, 0, 0), of(1, 0, 1), of(1, 0, 2));
+ DirtyPagesAndPartitions dirtyPages0 = createDirtyPagesAndPartitions(of(0, 0, 0), of(0, 0, 1));
+ DirtyPagesAndPartitions dirtyPages1 = createDirtyPagesAndPartitions(of(1, 0, 0), of(1, 0, 1), of(1, 0, 2));
assertEquals(0, EMPTY.dirtyPagesCount());
assertEquals(2, new CheckpointDirtyPages(List.of(dirtyPages0)).dirtyPagesCount());
@@ -61,18 +61,18 @@ public class CheckpointDirtyPagesTest extends BaseIgniteAbstractTest {
}
@Test
- void testToDirtyPageIdQueue() {
- assertTrue(EMPTY.toDirtyPageIdQueue().isEmpty());
+ void testToDirtyPartitionQueue() {
+ assertTrue(EMPTY.toDirtyPartitionQueue().isEmpty());
- DataRegionDirtyPages<FullPageId[]> dirtyPages0 = createDirtyPages(of(0, 0, 0));
- DataRegionDirtyPages<FullPageId[]> dirtyPages1 = createDirtyPages(of(1, 0, 0), of(1, 0, 1));
- DataRegionDirtyPages<FullPageId[]> dirtyPages2 = createDirtyPages(of(2, 0, 0), of(2, 1, 0), of(3, 2, 2));
+ DirtyPagesAndPartitions dirtyPages0 = createDirtyPagesAndPartitions(of(0, 0, 0));
+ DirtyPagesAndPartitions dirtyPages1 = createDirtyPagesAndPartitions(of(1, 0, 0), of(1, 0, 1));
+ DirtyPagesAndPartitions dirtyPages2 = createDirtyPagesAndPartitions(of(2, 0, 0), of(2, 1, 0), of(3, 2, 2));
- CheckpointDirtyPages checkpointDirtyPages = new CheckpointDirtyPages(List.of(dirtyPages0, dirtyPages1, dirtyPages2));
+ var checkpointDirtyPages = new CheckpointDirtyPages(List.of(dirtyPages0, dirtyPages1, dirtyPages2));
assertThat(
- toListPair(checkpointDirtyPages.toDirtyPageIdQueue()),
- equalTo(toListPair(dirtyPages0, dirtyPages1, dirtyPages2))
+ toListPair(checkpointDirtyPages.toDirtyPartitionQueue()),
+ equalTo(toListDirtyPartitionPair(dirtyPages0, dirtyPages1, dirtyPages2))
);
}
@@ -80,16 +80,16 @@ public class CheckpointDirtyPagesTest extends BaseIgniteAbstractTest {
void testGetPartitionViewByPageMemory() {
assertThrows(IllegalArgumentException.class, () -> EMPTY.getPartitionView(mock(PersistentPageMemory.class), 0, 0));
- DataRegionDirtyPages<FullPageId[]> dirtyPages0 = createDirtyPages(of(0, 0, 0));
- DataRegionDirtyPages<FullPageId[]> dirtyPages1 = createDirtyPages(of(5, 0, 0));
- DataRegionDirtyPages<FullPageId[]> dirtyPages2 = createDirtyPages(of(1, 0, 0), of(1, 0, 1));
- DataRegionDirtyPages<FullPageId[]> dirtyPages3 = createDirtyPages(
+ DirtyPagesAndPartitions dirtyPages0 = createDirtyPagesAndPartitions(of(0, 0, 0));
+ DirtyPagesAndPartitions dirtyPages1 = createDirtyPagesAndPartitions(of(5, 0, 0));
+ DirtyPagesAndPartitions dirtyPages2 = createDirtyPagesAndPartitions(of(1, 0, 0), of(1, 0, 1));
+ DirtyPagesAndPartitions dirtyPages3 = createDirtyPagesAndPartitions(
of(2, 0, 0), of(2, 0, 1),
of(2, 1, 1),
of(3, 2, 2), of(3, 2, 3)
);
- CheckpointDirtyPages checkpointDirtyPages = new CheckpointDirtyPages(List.of(dirtyPages0, dirtyPages1, dirtyPages2, dirtyPages3));
+ var checkpointDirtyPages = new CheckpointDirtyPages(List.of(dirtyPages0, dirtyPages1, dirtyPages2, dirtyPages3));
assertNull(checkpointDirtyPages.getPartitionView(dirtyPages0.pageMemory, 4, 0));
assertNull(checkpointDirtyPages.getPartitionView(dirtyPages1.pageMemory, 4, 0));
@@ -99,83 +99,36 @@ public class CheckpointDirtyPagesTest extends BaseIgniteAbstractTest {
assertNull(checkpointDirtyPages.getPartitionView(dirtyPages2.pageMemory, 5, 0));
assertThat(
- toListPair(checkpointDirtyPages.getPartitionView(dirtyPages0.pageMemory, 0, 0)),
- equalTo(toListPair(dirtyPages0))
+ toListDirtyPagePair(checkpointDirtyPages.getPartitionView(dirtyPages0.pageMemory, 0, 0)),
+ equalTo(toListDirtyPagePair(dirtyPages0))
);
assertThat(
- toListPair(checkpointDirtyPages.getPartitionView(dirtyPages1.pageMemory, 5, 0)),
- equalTo(toListPair(dirtyPages1))
+ toListDirtyPagePair(checkpointDirtyPages.getPartitionView(dirtyPages1.pageMemory, 5, 0)),
+ equalTo(toListDirtyPagePair(dirtyPages1))
);
assertThat(
- toListPair(checkpointDirtyPages.getPartitionView(dirtyPages2.pageMemory, 1, 0)),
- equalTo(toListPair(dirtyPages2))
+ toListDirtyPagePair(checkpointDirtyPages.getPartitionView(dirtyPages2.pageMemory, 1, 0)),
+ equalTo(toListDirtyPagePair(dirtyPages2))
);
assertThat(
- toListPair(checkpointDirtyPages.getPartitionView(dirtyPages3.pageMemory, 2, 0)),
- equalTo(toListPair(equalsByGroupAndPartition(2, 0), dirtyPages3))
+ toListDirtyPagePair(checkpointDirtyPages.getPartitionView(dirtyPages3.pageMemory, 2, 0)),
+ equalTo(toListDirtyPagePair(equalsByGroupAndPartition(2, 0), dirtyPages3))
);
assertThat(
- toListPair(checkpointDirtyPages.getPartitionView(dirtyPages3.pageMemory, 2, 1)),
- equalTo(toListPair(equalsByGroupAndPartition(2, 1), dirtyPages3))
+ toListDirtyPagePair(checkpointDirtyPages.getPartitionView(dirtyPages3.pageMemory, 2, 1)),
+ equalTo(toListDirtyPagePair(equalsByGroupAndPartition(2, 1), dirtyPages3))
);
assertThat(
- toListPair(checkpointDirtyPages.getPartitionView(dirtyPages3.pageMemory, 3, 2)),
- equalTo(toListPair(equalsByGroupAndPartition(3, 2), dirtyPages3))
+ toListDirtyPagePair(checkpointDirtyPages.getPartitionView(dirtyPages3.pageMemory, 3, 2)),
+ equalTo(toListDirtyPagePair(equalsByGroupAndPartition(3, 2), dirtyPages3))
);
}
- @Test
- void testNextPartitionView() {
- assertNull(EMPTY.nextPartitionView(null));
-
- DataRegionDirtyPages<FullPageId[]> dirtyPages0 = createDirtyPages(of(0, 0, 0));
- DataRegionDirtyPages<FullPageId[]> dirtyPages1 = createDirtyPages(of(5, 0, 0));
- DataRegionDirtyPages<FullPageId[]> dirtyPages2 = createDirtyPages(of(1, 0, 0), of(1, 0, 1));
- DataRegionDirtyPages<FullPageId[]> dirtyPages3 = createDirtyPages(
- of(2, 0, 0), of(2, 0, 1),
- of(2, 1, 1),
- of(3, 2, 2), of(3, 2, 3)
- );
-
- CheckpointDirtyPages checkpointDirtyPages = new CheckpointDirtyPages(List.of(dirtyPages0, dirtyPages1, dirtyPages2, dirtyPages3));
-
- CheckpointDirtyPagesView view = checkpointDirtyPages.nextPartitionView(null);
-
- assertThat(toListPair(view), equalTo(toListPair(dirtyPages0)));
-
- assertThat(
- toListPair(view = checkpointDirtyPages.nextPartitionView(view)),
- equalTo(toListPair(dirtyPages1))
- );
-
- assertThat(
- toListPair(view = checkpointDirtyPages.nextPartitionView(view)),
- equalTo(toListPair(dirtyPages2))
- );
-
- assertThat(
- toListPair(view = checkpointDirtyPages.nextPartitionView(view)),
- equalTo(toListPair(equalsByGroupAndPartition(2, 0), dirtyPages3))
- );
-
- assertThat(
- toListPair(view = checkpointDirtyPages.nextPartitionView(view)),
- equalTo(toListPair(equalsByGroupAndPartition(2, 1), dirtyPages3))
- );
-
- assertThat(
- toListPair(view = checkpointDirtyPages.nextPartitionView(view)),
- equalTo(toListPair(equalsByGroupAndPartition(3, 2), dirtyPages3))
- );
-
- assertNull(checkpointDirtyPages.nextPartitionView(view));
- }
-
@Test
void testSortDirtyPageIds() {
List<FullPageId> pageIds = new ArrayList<>(List.of(
@@ -196,10 +149,8 @@ public class CheckpointDirtyPagesTest extends BaseIgniteAbstractTest {
);
}
- private static DataRegionDirtyPages<FullPageId[]> createDirtyPages(FullPageId... pageIds) {
- Arrays.sort(pageIds, DIRTY_PAGE_COMPARATOR);
-
- return new DataRegionDirtyPages<>(mock(PersistentPageMemory.class), pageIds);
+ private static DirtyPagesAndPartitions createDirtyPagesAndPartitions(FullPageId... pageIds) {
+ return TestCheckpointUtils.createDirtyPagesAndPartitions(mock(PersistentPageMemory.class), pageIds);
}
private static FullPageId of(int groupId, int partId, int pageIdx) {
@@ -220,13 +171,21 @@ public class CheckpointDirtyPagesTest extends BaseIgniteAbstractTest {
.collect(toList());
}
- private static List<IgniteBiTuple<PersistentPageMemory, FullPageId>> toListPair(DataRegionDirtyPages<FullPageId[]>... dirtyPages) {
- return toListPair(dirtyPageId -> true, dirtyPages);
+ private static List<IgniteBiTuple<PersistentPageMemory, GroupPartitionId>> toListDirtyPartitionPair(
+ DirtyPagesAndPartitions... dirtyPagesAndPartitions
+ ) {
+ return Stream.of(dirtyPagesAndPartitions)
+ .flatMap(pages -> pages.dirtyPartitions.stream().map(partitionId -> new IgniteBiTuple<>(pages.pageMemory, partitionId)))
+ .collect(toList());
+ }
+
+ private static List<IgniteBiTuple<PersistentPageMemory, FullPageId>> toListDirtyPagePair(DirtyPagesAndPartitions... dirtyPages) {
+ return toListDirtyPagePair(dirtyPageId -> true, dirtyPages);
}
- private static List<IgniteBiTuple<PersistentPageMemory, FullPageId>> toListPair(
+ private static List<IgniteBiTuple<PersistentPageMemory, FullPageId>> toListDirtyPagePair(
Predicate<FullPageId> predicate,
- DataRegionDirtyPages<FullPageId[]>... dirtyPages
+ DirtyPagesAndPartitions... dirtyPages
) {
return Stream.of(dirtyPages)
.flatMap(pages -> Stream.of(pages.dirtyPages)
@@ -236,7 +195,7 @@ public class CheckpointDirtyPagesTest extends BaseIgniteAbstractTest {
.collect(toList());
}
- private static List<IgniteBiTuple<PersistentPageMemory, FullPageId>> toListPair(CheckpointDirtyPagesView view) {
+ private static List<IgniteBiTuple<PersistentPageMemory, FullPageId>> toListDirtyPagePair(CheckpointDirtyPagesView view) {
return IntStream.range(0, view.size()).mapToObj(i -> new IgniteBiTuple<>(view.pageMemory(), view.get(i))).collect(toList());
}
diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointManagerTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointManagerTest.java
index 76d6dd29b6..d21fcfc89a 100644
--- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointManagerTest.java
+++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointManagerTest.java
@@ -23,6 +23,7 @@ import static org.apache.ignite.internal.pagememory.persistence.CheckpointUrgenc
import static org.apache.ignite.internal.pagememory.persistence.CheckpointUrgency.SHOULD_TRIGGER;
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointManager.checkpointUrgency;
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointManager.pageIndexesForDeltaFilePageStore;
+import static org.apache.ignite.internal.pagememory.persistence.checkpoint.TestCheckpointUtils.createDirtyPagesAndPartitions;
import static org.apache.ignite.internal.pagememory.util.PageIdUtils.pageId;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
@@ -146,9 +147,9 @@ public class CheckpointManagerTest extends BaseIgniteAbstractTest {
PersistentPageMemory pageMemory0 = mock(PersistentPageMemory.class);
PersistentPageMemory pageMemory1 = mock(PersistentPageMemory.class);
- CheckpointDirtyPages dirtyPages = new CheckpointDirtyPages(List.of(
- new DataRegionDirtyPages<>(pageMemory0, dirtyPageArray(0, 0, 1)),
- new DataRegionDirtyPages<>(pageMemory1, dirtyPageArray(0, 1, 2, 3, 4))
+ var dirtyPages = new CheckpointDirtyPages(List.of(
+ createDirtyPagesAndPartitions(pageMemory0, dirtyPageArray(0, 0, 1)),
+ createDirtyPagesAndPartitions(pageMemory1, dirtyPageArray(0, 1, 2, 3, 4))
));
assertArrayEquals(new int[]{0, 1}, pageIndexesForDeltaFilePageStore(dirtyPages.getPartitionView(pageMemory0, 0, 0)));
@@ -160,9 +161,9 @@ public class CheckpointManagerTest extends BaseIgniteAbstractTest {
PersistentPageMemory pageMemory0 = mock(PersistentPageMemory.class);
PersistentPageMemory pageMemory1 = mock(PersistentPageMemory.class);
- CheckpointDirtyPages dirtyPages = new CheckpointDirtyPages(List.of(
- new DataRegionDirtyPages<>(pageMemory0, dirtyPageArray(0, 0, 0, 1)),
- new DataRegionDirtyPages<>(pageMemory1, dirtyPageArray(0, 1, 0, 2, 3, 4))
+ var dirtyPages = new CheckpointDirtyPages(List.of(
+ createDirtyPagesAndPartitions(pageMemory0, dirtyPageArray(0, 0, 0, 1)),
+ createDirtyPagesAndPartitions(pageMemory1, dirtyPageArray(0, 1, 0, 2, 3, 4))
));
assertArrayEquals(new int[]{0, 1}, pageIndexesForDeltaFilePageStore(dirtyPages.getPartitionView(pageMemory0, 0, 0)));
@@ -209,9 +210,7 @@ public class CheckpointManagerTest extends BaseIgniteAbstractTest {
CheckpointProgress checkpointProgress = mock(CheckpointProgress.class);
- CheckpointDirtyPages dirtyPages = new CheckpointDirtyPages(List.of(
- new DataRegionDirtyPages<>(pageMemory, new FullPageId[]{dirtyPageId})
- ));
+ var dirtyPages = new CheckpointDirtyPages(List.of(createDirtyPagesAndPartitions(pageMemory, dirtyPageId)));
when(checkpointProgress.inProgress()).thenReturn(true);
diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriterTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriterTest.java
index 97d5dfdb04..94c305aac6 100644
--- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriterTest.java
+++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriterTest.java
@@ -22,6 +22,7 @@ import static org.apache.ignite.internal.pagememory.PageIdAllocator.FLAG_AUX;
import static org.apache.ignite.internal.pagememory.PageIdAllocator.FLAG_DATA;
import static org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory.TRY_AGAIN_TAG;
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointTestUtils.createPartitionMetaManager;
+import static org.apache.ignite.internal.pagememory.persistence.checkpoint.TestCheckpointUtils.createDirtyPagesAndPartitions;
import static org.apache.ignite.internal.pagememory.util.PageIdUtils.pageId;
import static org.apache.ignite.internal.util.GridUnsafe.allocateBuffer;
import static org.apache.ignite.internal.util.GridUnsafe.bufferAddress;
@@ -101,9 +102,9 @@ public class CheckpointPagesWriterTest extends BaseIgniteAbstractTest {
FullPageId fullPageId5 = new FullPageId(pageId(0, FLAG_DATA, 5), 0);
FullPageId fullPageId6 = new FullPageId(pageId(1, FLAG_DATA, 6), 0);
- IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> writePageIds = new IgniteConcurrentMultiPairQueue<>(
- Map.of(pageMemory, List.of(fullPageId1, fullPageId2, fullPageId3, fullPageId4, fullPageId5, fullPageId6))
- );
+ CheckpointDirtyPages checkpointDirtyPages = new CheckpointDirtyPages(List.of(
+ createDirtyPagesAndPartitions(pageMemory, fullPageId1, fullPageId2, fullPageId3, fullPageId4, fullPageId5, fullPageId6)
+ ));
GroupPartitionId groupPartId0 = groupPartId(0, 0);
GroupPartitionId groupPartId1 = groupPartId(0, 1);
@@ -123,13 +124,17 @@ public class CheckpointPagesWriterTest extends BaseIgniteAbstractTest {
CheckpointMetricsTracker tracker = new CheckpointMetricsTracker();
CheckpointProgressImpl progressImpl = new CheckpointProgressImpl(0);
+ progressImpl.pagesToWrite(checkpointDirtyPages);
PartitionMeta partitionMeta0 = mock(PartitionMeta.class);
PartitionMeta partitionMeta1 = mock(PartitionMeta.class);
+ IgniteConcurrentMultiPairQueue<PersistentPageMemory, GroupPartitionId> dirtyPartitionQueue
+ = checkpointDirtyPages.toDirtyPartitionQueue();
+
CheckpointPagesWriter pagesWriter = new CheckpointPagesWriter(
tracker,
- writePageIds,
+ dirtyPartitionQueue,
singletonList(pageMemory),
updatedPartitions,
doneFuture,
@@ -148,7 +153,7 @@ public class CheckpointPagesWriterTest extends BaseIgniteAbstractTest {
assertDoesNotThrow(() -> doneFuture.get(1, TimeUnit.SECONDS));
- assertTrue(writePageIds.isEmpty());
+ assertTrue(dirtyPartitionQueue.isEmpty());
assertThat(updatedPartitions.keySet(), containsInAnyOrder(groupPartId0, groupPartId1));
@@ -162,17 +167,20 @@ public class CheckpointPagesWriterTest extends BaseIgniteAbstractTest {
writtenFullPageIds.getAllValues(),
equalTo(List.of(
// At the beginning, we write the partition meta for each new partition.
- fullPageId(0, 0, 0),
+ new FullPageId(pageId(0, FLAG_AUX, 0), 0),
// Order is different because the first 3 pages we have to try to write to the page store 2 times.
fullPageId4, fullPageId5,
- fullPageId(0, 1, 0),
- fullPageId6, fullPageId1, fullPageId2, fullPageId3
+ // At the beginning, we write the partition meta for each new partition.
+ new FullPageId(pageId(1, FLAG_AUX, 0), 0),
+ fullPageId6,
+ // Now the retry pages.
+ fullPageId1, fullPageId2, fullPageId3
))
);
- verify(beforePageWrite, times(11)).run();
+ verify(beforePageWrite, times(14)).run();
- verify(threadBuf, times(2)).get();
+ verify(threadBuf, times(1)).get();
verify(partitionMeta0, times(1)).metaSnapshot(any(UUID.class));
verify(partitionMeta1, times(1)).metaSnapshot(any(UUID.class));
@@ -195,15 +203,22 @@ public class CheckpointPagesWriterTest extends BaseIgniteAbstractTest {
GroupPartitionId groupPartId = groupPartId(0, 0);
+ CheckpointDirtyPages checkpointDirtyPages = new CheckpointDirtyPages(List.of(
+ createDirtyPagesAndPartitions(pageMemory, new FullPageId(pageId(0, FLAG_DATA, 1), 0))
+ ));
+
+ CheckpointProgressImpl checkpointProgress = new CheckpointProgressImpl(0);
+ checkpointProgress.pagesToWrite(checkpointDirtyPages);
+
CheckpointPagesWriter pagesWriter = new CheckpointPagesWriter(
new CheckpointMetricsTracker(),
- new IgniteConcurrentMultiPairQueue<>(Map.of(pageMemory, List.of(fullPageId(0, 0, 1)))),
+ new IgniteConcurrentMultiPairQueue<>(Map.of(pageMemory, List.of(new GroupPartitionId(0, 0)))),
singletonList(pageMemory),
new ConcurrentHashMap<>(),
doneFuture,
() -> {},
createThreadLocalBuffer(),
- new CheckpointProgressImpl(0),
+ checkpointProgress,
createDirtyPageWriter(null),
ioRegistry,
createPartitionMetaManager(Map.of(groupPartId, mock(PartitionMeta.class))),
@@ -238,23 +253,29 @@ public class CheckpointPagesWriterTest extends BaseIgniteAbstractTest {
any(CheckpointMetricsTracker.class)
);
- IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> writePageIds = new IgniteConcurrentMultiPairQueue<>(
- Map.of(pageMemory, List.of(fullPageId(0, 0, 1), fullPageId(0, 0, 2)))
- );
+ CheckpointDirtyPages checkpointDirtyPages = new CheckpointDirtyPages(List.of(
+ createDirtyPagesAndPartitions(pageMemory, fullPageId(0, 0, 1), fullPageId(0, 1, 2))
+ ));
GroupPartitionId groupPartId = groupPartId(0, 0);
ConcurrentMap<GroupPartitionId, LongAdder> updatedPartitions = new ConcurrentHashMap<>();
+ CheckpointProgressImpl checkpointProgress = new CheckpointProgressImpl(0);
+ checkpointProgress.pagesToWrite(checkpointDirtyPages);
+
+ IgniteConcurrentMultiPairQueue<PersistentPageMemory, GroupPartitionId> dirtyPartitionQueue
+ = checkpointDirtyPages.toDirtyPartitionQueue();
+
CheckpointPagesWriter pagesWriter = new CheckpointPagesWriter(
new CheckpointMetricsTracker(),
- writePageIds,
+ dirtyPartitionQueue,
singletonList(pageMemory),
updatedPartitions,
doneFuture,
() -> {},
createThreadLocalBuffer(),
- new CheckpointProgressImpl(0),
+ checkpointProgress,
createDirtyPageWriter(null),
ioRegistry,
createPartitionMetaManager(Map.of(groupPartId, mock(PartitionMeta.class))),
@@ -265,7 +286,7 @@ public class CheckpointPagesWriterTest extends BaseIgniteAbstractTest {
assertDoesNotThrow(() -> doneFuture.get(1, TimeUnit.SECONDS));
- assertThat(writePageIds.size(), equalTo(1));
+ assertThat(dirtyPartitionQueue.size(), equalTo(1));
assertThat(updatedPartitions.keySet(), contains(groupPartId));
}
diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointTest.java
index 6f5f6c3fc5..9facbb11de 100644
--- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointTest.java
+++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.pagememory.persistence.checkpoint;
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.EMPTY;
+import static org.apache.ignite.internal.pagememory.persistence.checkpoint.TestCheckpointUtils.createDirtyPagesAndPartitions;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
@@ -38,11 +39,11 @@ public class CheckpointTest extends BaseIgniteAbstractTest {
assertFalse(new Checkpoint(EMPTY, progress).hasDelta());
- DataRegionDirtyPages<FullPageId[]> biTuple = new DataRegionDirtyPages<>(
+ DirtyPagesAndPartitions dirtyPagesAndPartitions = createDirtyPagesAndPartitions(
mock(PersistentPageMemory.class),
- new FullPageId[]{new FullPageId(0, 1)}
+ new FullPageId(0, 1)
);
- assertTrue(new Checkpoint(new CheckpointDirtyPages(List.of(biTuple)), progress).hasDelta());
+ assertTrue(new Checkpoint(new CheckpointDirtyPages(List.of(dirtyPagesAndPartitions)), progress).hasDelta());
}
}
diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointWorkflowTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointWorkflowTest.java
index 7f670a4efa..9e45b41e4b 100644
--- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointWorkflowTest.java
+++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointWorkflowTest.java
@@ -33,6 +33,7 @@ import static org.apache.ignite.internal.pagememory.persistence.checkpoint.Check
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointWorkflowTest.TestCheckpointListener.BEFORE_CHECKPOINT_BEGIN;
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointWorkflowTest.TestCheckpointListener.ON_CHECKPOINT_BEGIN;
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointWorkflowTest.TestCheckpointListener.ON_MARK_CHECKPOINT_BEGIN;
+import static org.apache.ignite.internal.pagememory.persistence.checkpoint.TestCheckpointUtils.createDirtyPagesAndPartitions;
import static org.apache.ignite.internal.pagememory.util.PageIdUtils.pageId;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.runAsync;
@@ -314,7 +315,7 @@ public class CheckpointWorkflowTest extends BaseIgniteAbstractTest {
verify(progressImpl, times(1)).pagesToWrite(any(CheckpointDirtyPages.class));
verify(progressImpl, times(1)).initCounters(anyInt());
- CheckpointDirtyPagesView dirtyPagesView = checkpoint.dirtyPages.nextPartitionView(null);
+ CheckpointDirtyPagesView dirtyPagesView = checkpoint.dirtyPages.getPartitionView(pageMemory, 0, 0);
assertThat(toListDirtyPageIds(dirtyPagesView), equalTo(dirtyPages));
assertThat(dirtyPagesView.pageMemory(), equalTo(dataRegion.pageMemory()));
@@ -383,7 +384,7 @@ public class CheckpointWorkflowTest extends BaseIgniteAbstractTest {
workflow.addCheckpointListener(checkpointListener, dataRegion);
workflow.markCheckpointEnd(new Checkpoint(
- new CheckpointDirtyPages(List.of(createCheckpointDirtyPages(pageMemory, of(0, 0, 0)))),
+ new CheckpointDirtyPages(List.of(createDirtyPagesAndPartitions(pageMemory, of(0, 0, 0)))),
progressImpl
));
@@ -408,15 +409,18 @@ public class CheckpointWorkflowTest extends BaseIgniteAbstractTest {
@Test
void testCreateAndSortCheckpointDirtyPages() throws Exception {
+ PersistentPageMemory pageMemory0 = mock(PersistentPageMemory.class);
+ PersistentPageMemory pageMemory1 = mock(PersistentPageMemory.class);
+
DataRegionDirtyPages<Collection<FullPageId>> dataRegionDirtyPages0 = createDataRegionDirtyPages(
- mock(PersistentPageMemory.class),
+ pageMemory0,
of(10, 10, 2), of(10, 10, 1), of(10, 10, 0),
of(10, 5, 100), of(10, 5, 99),
of(10, 1, 50), of(10, 1, 51), of(10, 1, 99)
);
DataRegionDirtyPages<Collection<FullPageId>> dataRegionDirtyPages1 = createDataRegionDirtyPages(
- mock(PersistentPageMemory.class),
+ pageMemory1,
of(77, 5, 100), of(77, 5, 99),
of(88, 1, 51), of(88, 1, 50), of(88, 1, 99),
of(66, 33, 0), of(66, 33, 1), of(66, 33, 2)
@@ -435,32 +439,32 @@ public class CheckpointWorkflowTest extends BaseIgniteAbstractTest {
new DataRegionsDirtyPages(List.of(dataRegionDirtyPages0, dataRegionDirtyPages1))
);
- CheckpointDirtyPagesView dirtyPagesView = sortCheckpointDirtyPages.nextPartitionView(null);
+ CheckpointDirtyPagesView dirtyPagesView = sortCheckpointDirtyPages.getPartitionView(pageMemory0, 10, 1);
assertThat(toListDirtyPageIds(dirtyPagesView), equalTo(List.of(of(10, 1, 50), of(10, 1, 51), of(10, 1, 99))));
assertThat(dirtyPagesView.pageMemory(), equalTo(dataRegionDirtyPages0.pageMemory));
- dirtyPagesView = sortCheckpointDirtyPages.nextPartitionView(dirtyPagesView);
+ dirtyPagesView = sortCheckpointDirtyPages.getPartitionView(pageMemory0, 10, 5);
assertThat(toListDirtyPageIds(dirtyPagesView), equalTo(List.of(of(10, 5, 99), of(10, 5, 100))));
assertThat(dirtyPagesView.pageMemory(), equalTo(dataRegionDirtyPages0.pageMemory));
- dirtyPagesView = sortCheckpointDirtyPages.nextPartitionView(dirtyPagesView);
+ dirtyPagesView = sortCheckpointDirtyPages.getPartitionView(pageMemory0, 10, 10);
assertThat(toListDirtyPageIds(dirtyPagesView), equalTo(List.of(of(10, 10, 0), of(10, 10, 1), of(10, 10, 2))));
assertThat(dirtyPagesView.pageMemory(), equalTo(dataRegionDirtyPages0.pageMemory));
- dirtyPagesView = sortCheckpointDirtyPages.nextPartitionView(dirtyPagesView);
+ dirtyPagesView = sortCheckpointDirtyPages.getPartitionView(pageMemory1, 66, 33);
assertThat(toListDirtyPageIds(dirtyPagesView), equalTo(List.of(of(66, 33, 0), of(66, 33, 1), of(66, 33, 2))));
assertThat(dirtyPagesView.pageMemory(), equalTo(dataRegionDirtyPages1.pageMemory));
- dirtyPagesView = sortCheckpointDirtyPages.nextPartitionView(dirtyPagesView);
+ dirtyPagesView = sortCheckpointDirtyPages.getPartitionView(pageMemory1, 77, 5);
assertThat(toListDirtyPageIds(dirtyPagesView), equalTo(List.of(of(77, 5, 99), of(77, 5, 100))));
assertThat(dirtyPagesView.pageMemory(), equalTo(dataRegionDirtyPages1.pageMemory));
- dirtyPagesView = sortCheckpointDirtyPages.nextPartitionView(dirtyPagesView);
+ dirtyPagesView = sortCheckpointDirtyPages.getPartitionView(pageMemory1, 88, 1);
assertThat(toListDirtyPageIds(dirtyPagesView), equalTo(List.of(of(88, 1, 50), of(88, 1, 51), of(88, 1, 99))));
assertThat(dirtyPagesView.pageMemory(), equalTo(dataRegionDirtyPages1.pageMemory));
@@ -482,16 +486,19 @@ public class CheckpointWorkflowTest extends BaseIgniteAbstractTest {
workflow.start();
+ PersistentPageMemory pageMemory0 = mock(PersistentPageMemory.class);
+ PersistentPageMemory pageMemory1 = mock(PersistentPageMemory.class);
+
CheckpointDirtyPages sortCheckpointDirtyPages = workflow.createAndSortCheckpointDirtyPages(new DataRegionsDirtyPages(List.of(
- createDataRegionDirtyPages(mock(PersistentPageMemory.class), dirtyPages1),
- createDataRegionDirtyPages(mock(PersistentPageMemory.class), dirtyPages0)
+ createDataRegionDirtyPages(pageMemory1, dirtyPages1),
+ createDataRegionDirtyPages(pageMemory0, dirtyPages0)
)));
- CheckpointDirtyPagesView dirtyPagesView = sortCheckpointDirtyPages.nextPartitionView(null);
+ CheckpointDirtyPagesView dirtyPagesView = sortCheckpointDirtyPages.getPartitionView(pageMemory1, 1, 1);
assertThat(toListDirtyPageIds(dirtyPagesView), equalTo(List.of(dirtyPages1)));
- dirtyPagesView = sortCheckpointDirtyPages.nextPartitionView(dirtyPagesView);
+ dirtyPagesView = sortCheckpointDirtyPages.getPartitionView(pageMemory0, 0, 0);
assertThat(
toListDirtyPageIds(dirtyPagesView),
@@ -533,7 +540,7 @@ public class CheckpointWorkflowTest extends BaseIgniteAbstractTest {
assertEquals(1, checkpoint.dirtyPagesSize);
- CheckpointDirtyPagesView dirtyPagesView = checkpoint.dirtyPages.nextPartitionView(null);
+ CheckpointDirtyPagesView dirtyPagesView = checkpoint.dirtyPages.getPartitionView(pageMemory, groupId, partitionId);
assertNotNull(dirtyPagesView);
assertThat(toListDirtyPageIds(dirtyPagesView), is(List.of(metaPageId)));
@@ -576,7 +583,7 @@ public class CheckpointWorkflowTest extends BaseIgniteAbstractTest {
assertEquals(2, checkpoint.dirtyPagesSize);
- CheckpointDirtyPagesView dirtyPagesView = checkpoint.dirtyPages.nextPartitionView(null);
+ CheckpointDirtyPagesView dirtyPagesView = checkpoint.dirtyPages.getPartitionView(pageMemory, groupId, partitionId);
assertNotNull(dirtyPagesView);
assertThat(toListDirtyPageIds(dirtyPagesView), is(List.of(metaPageId, dataPageId)));
@@ -663,13 +670,6 @@ public class CheckpointWorkflowTest extends BaseIgniteAbstractTest {
return mock;
}
- private static DataRegionDirtyPages<FullPageId[]> createCheckpointDirtyPages(
- PersistentPageMemory pageMemory,
- FullPageId... pageIds
- ) {
- return new DataRegionDirtyPages<>(pageMemory, pageIds);
- }
-
private static DataRegionDirtyPages<Collection<FullPageId>> createDataRegionDirtyPages(
PersistentPageMemory pageMemory,
FullPageId... pageIds
diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointerTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointerTest.java
index 38506b649d..976174068a 100644
--- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointerTest.java
+++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointerTest.java
@@ -22,11 +22,11 @@ import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.ignite.internal.pagememory.persistence.FakePartitionMeta.FACTORY;
-import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.DIRTY_PAGE_COMPARATOR;
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.EMPTY;
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointState.FINISHED;
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointState.LOCK_RELEASED;
import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointState.LOCK_TAKEN;
+import static org.apache.ignite.internal.pagememory.persistence.checkpoint.TestCheckpointUtils.createDirtyPagesAndPartitions;
import static org.apache.ignite.internal.pagememory.util.PageIdUtils.pageId;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.runAsync;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
@@ -56,7 +56,6 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -391,7 +390,7 @@ public class CheckpointerTest extends BaseIgniteAbstractTest {
assertDoesNotThrow(checkpointer::doCheckpoint);
- verify(dirtyPages, times(1)).toDirtyPageIdQueue();
+ verify(dirtyPages, times(1)).toDirtyPartitionQueue();
verify(checkpointer, times(1)).startCheckpointProgress();
verify(compactor, times(1)).triggerCompaction();
verify(mockLogSyncer, times(1)).sync();
@@ -423,7 +422,7 @@ public class CheckpointerTest extends BaseIgniteAbstractTest {
assertDoesNotThrow(checkpointer::doCheckpoint);
- verify(dirtyPages, never()).toDirtyPageIdQueue();
+ verify(dirtyPages, never()).toDirtyPartitionQueue();
verify(checkpointer, times(1)).startCheckpointProgress();
verify(compactor, never()).triggerCompaction();
@@ -518,10 +517,8 @@ public class CheckpointerTest extends BaseIgniteAbstractTest {
onPartitionDestructionFuture.get(1, SECONDS);
}
- private CheckpointDirtyPages dirtyPages(PersistentPageMemory pageMemory, FullPageId... pageIds) {
- Arrays.sort(pageIds, DIRTY_PAGE_COMPARATOR);
-
- return new CheckpointDirtyPages(List.of(new DataRegionDirtyPages<>(pageMemory, pageIds)));
+ private static CheckpointDirtyPages dirtyPages(PersistentPageMemory pageMemory, FullPageId... pageIds) {
+ return new CheckpointDirtyPages(List.of(createDirtyPagesAndPartitions(pageMemory, pageIds)));
}
private CheckpointWorkflow createCheckpointWorkflow(CheckpointDirtyPages dirtyPages) throws Exception {
diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/TestCheckpointUtils.java
similarity index 54%
copy from modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointTest.java
copy to modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/TestCheckpointUtils.java
index 6f5f6c3fc5..217e19b01f 100644
--- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointTest.java
+++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/TestCheckpointUtils.java
@@ -17,32 +17,24 @@
package org.apache.ignite.internal.pagememory.persistence.checkpoint;
-import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.EMPTY;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.mock;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.DIRTY_PAGE_COMPARATOR;
-import java.util.List;
+import java.util.Arrays;
import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
-import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
-import org.junit.jupiter.api.Test;
-/**
- * For {@link Checkpoint} testing.
- */
-public class CheckpointTest extends BaseIgniteAbstractTest {
- @Test
- void testHasDelta() {
- CheckpointProgressImpl progress = mock(CheckpointProgressImpl.class);
-
- assertFalse(new Checkpoint(EMPTY, progress).hasDelta());
+/** Helper class for checkpoint testing that may contain useful methods and constants. */
+class TestCheckpointUtils {
+ /** Sorts dirty pages and creates a new instance {@link DirtyPagesAndPartitions}. */
+ static DirtyPagesAndPartitions createDirtyPagesAndPartitions(PersistentPageMemory pageMemory, FullPageId... dirtyPages) {
+ Arrays.sort(dirtyPages, DIRTY_PAGE_COMPARATOR);
- DataRegionDirtyPages<FullPageId[]> biTuple = new DataRegionDirtyPages<>(
- mock(PersistentPageMemory.class),
- new FullPageId[]{new FullPageId(0, 1)}
+ return new DirtyPagesAndPartitions(
+ pageMemory,
+ dirtyPages,
+ Arrays.stream(dirtyPages).map(GroupPartitionId::convert).collect(toSet())
);
-
- assertTrue(new Checkpoint(new CheckpointDirtyPages(List.of(biTuple)), progress).hasDelta());
}
}