rpuch commented on code in PR #907:
URL: https://github.com/apache/ignite-3/pull/907#discussion_r914472240
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemory.java:
##########
@@ -2095,6 +2095,7 @@ public Collection<FullPageId>
beginCheckpoint(CompletableFuture<?> allowToReplac
safeToUpdate.set(true);
+ // Less memory and no need for Set features.
Review Comment:
This comment seems to compare the current code with some alternative. But
where is the alternative?
I suggest to either elaborate on what we compare the current code with, or
remove the comment to avoid confusion.
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
+ static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
+ .comparingInt(FullPageId::groupId)
+ .thenComparingLong(FullPageId::effectivePageId);
+
+ /** Empty checkpoint dirty pages. */
+ static final CheckpointDirtyPages EMPTY = new
CheckpointDirtyPages(List.of());
+
+ /** Sorted dirty pages from data regions by groupId -> partitionId ->
pageIdx. */
+ private final List<IgniteBiTuple<PersistentPageMemory, List<FullPageId>>>
dirtyPages;
+
+ /** Total number of dirty pages. */
+ private final int dirtyPagesCount;
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(Map<PersistentPageMemory, List<FullPageId>>
dirtyPages) {
+ this(dirtyPages.isEmpty() ? List.of()
Review Comment:
Do we actually need this tiny economy? We seem to save a couple of
allocations each checkpoint, and checkpoints happen pretty rarely. This does
not seem worth complicating the code with this conditional operator.
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
+ static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
+ .comparingInt(FullPageId::groupId)
+ .thenComparingLong(FullPageId::effectivePageId);
+
+ /** Empty checkpoint dirty pages. */
+ static final CheckpointDirtyPages EMPTY = new
CheckpointDirtyPages(List.of());
+
+ /** Sorted dirty pages from data regions by groupId -> partitionId ->
pageIdx. */
+ private final List<IgniteBiTuple<PersistentPageMemory, List<FullPageId>>>
dirtyPages;
+
+ /** Total number of dirty pages. */
+ private final int dirtyPagesCount;
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(Map<PersistentPageMemory, List<FullPageId>>
dirtyPages) {
+ this(dirtyPages.isEmpty() ? List.of()
+ : dirtyPages.entrySet().stream().map(e -> new
IgniteBiTuple<>(e.getKey(), e.getValue())).collect(toList()));
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(List<IgniteBiTuple<PersistentPageMemory,
List<FullPageId>>> dirtyPages) {
+ assert dirtyPages instanceof RandomAccess : dirtyPages;
+
+ this.dirtyPages = dirtyPages;
+
+ int count = 0;
+
+ for (IgniteBiTuple<PersistentPageMemory, List<FullPageId>> pages :
dirtyPages) {
+ assert !pages.getValue().isEmpty() : pages.getKey();
+ assert pages.getValue() instanceof RandomAccess : pages.getValue();
+
+ count += pages.getValue().size();
+ }
+
+ dirtyPagesCount = count;
+ }
+
+ /**
+ * Returns total number of dirty pages.
+ */
+ public int dirtyPagesCount() {
+ return dirtyPagesCount;
+ }
+
+ /**
+ * Returns a queue of dirty pages to be written to a checkpoint.
+ */
+ public CheckpointDirtyPagesQueue toQueue() {
+ return new CheckpointDirtyPagesQueue();
+ }
+
+ /**
+ * Looks for dirty page views for a specific group and partition.
+ *
+ * @param grpId Group ID.
+ * @param partId Partition ID.
+ */
+ public @Nullable CheckpointDirtyPagesView findView(int grpId, int partId) {
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ FullPageId startPageId = new FullPageId(pageId(partId, (byte) 0, 0),
grpId);
+ FullPageId endPageId = new FullPageId(pageId(partId + 1, (byte) 0, 0),
grpId);
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ List<FullPageId> pageIds = dirtyPages.get(i).getValue();
+
+ int fromIndex = binarySearch(pageIds, startPageId,
DIRTY_PAGE_COMPARATOR);
+
+ fromIndex = fromIndex >= 0 ? fromIndex : Math.min(pageIds.size() -
1, -fromIndex - 1);
+
+ if (!equalsByGroupAndPartition(startPageId,
pageIds.get(fromIndex))) {
+ continue;
+ }
+
+ int toIndex = binarySearch(pageIds.subList(fromIndex,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toIndex = toIndex > 0 ? toIndex - 1 : -toIndex - 2;
Review Comment:
I suggest adding a comment that `toIndex` (before reassignment) cannot be 0,
to make the reassignment expression a little bit easier to understand
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
+ static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
+ .comparingInt(FullPageId::groupId)
+ .thenComparingLong(FullPageId::effectivePageId);
+
+ /** Empty checkpoint dirty pages. */
+ static final CheckpointDirtyPages EMPTY = new
CheckpointDirtyPages(List.of());
+
+ /** Sorted dirty pages from data regions by groupId -> partitionId ->
pageIdx. */
+ private final List<IgniteBiTuple<PersistentPageMemory, List<FullPageId>>>
dirtyPages;
+
+ /** Total number of dirty pages. */
+ private final int dirtyPagesCount;
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(Map<PersistentPageMemory, List<FullPageId>>
dirtyPages) {
+ this(dirtyPages.isEmpty() ? List.of()
+ : dirtyPages.entrySet().stream().map(e -> new
IgniteBiTuple<>(e.getKey(), e.getValue())).collect(toList()));
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(List<IgniteBiTuple<PersistentPageMemory,
List<FullPageId>>> dirtyPages) {
+ assert dirtyPages instanceof RandomAccess : dirtyPages;
+
+ this.dirtyPages = dirtyPages;
+
+ int count = 0;
+
+ for (IgniteBiTuple<PersistentPageMemory, List<FullPageId>> pages :
dirtyPages) {
+ assert !pages.getValue().isEmpty() : pages.getKey();
+ assert pages.getValue() instanceof RandomAccess : pages.getValue();
+
+ count += pages.getValue().size();
+ }
+
+ dirtyPagesCount = count;
+ }
+
+ /**
+ * Returns total number of dirty pages.
+ */
+ public int dirtyPagesCount() {
+ return dirtyPagesCount;
+ }
+
+ /**
+ * Returns a queue of dirty pages to be written to a checkpoint.
+ */
+ public CheckpointDirtyPagesQueue toQueue() {
+ return new CheckpointDirtyPagesQueue();
+ }
+
+ /**
+ * Looks for dirty page views for a specific group and partition.
+ *
+ * @param grpId Group ID.
+ * @param partId Partition ID.
+ */
+ public @Nullable CheckpointDirtyPagesView findView(int grpId, int partId) {
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ FullPageId startPageId = new FullPageId(pageId(partId, (byte) 0, 0),
grpId);
+ FullPageId endPageId = new FullPageId(pageId(partId + 1, (byte) 0, 0),
grpId);
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ List<FullPageId> pageIds = dirtyPages.get(i).getValue();
+
+ int fromIndex = binarySearch(pageIds, startPageId,
DIRTY_PAGE_COMPARATOR);
+
+ fromIndex = fromIndex >= 0 ? fromIndex : Math.min(pageIds.size() -
1, -fromIndex - 1);
+
+ if (!equalsByGroupAndPartition(startPageId,
pageIds.get(fromIndex))) {
+ continue;
+ }
+
+ int toIndex = binarySearch(pageIds.subList(fromIndex,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toIndex = toIndex > 0 ? toIndex - 1 : -toIndex - 2;
+
+ return new CheckpointDirtyPagesView(i, fromIndex, fromIndex +
toIndex);
+ }
+
+ return null;
+ }
+
+ /**
+ * Looks for the next dirty page 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 nextView(@Nullable
CheckpointDirtyPagesView currentView) {
+ assert currentView == null || currentView.owner() == this :
currentView;
+
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ int index;
+ int fromPosition;
+
+ if (currentView == null) {
+ index = 0;
+ fromPosition = 0;
+ } else {
+ index = currentView.isToPositionLast() ? currentView.index + 1 :
currentView.index;
+ fromPosition = currentView.isToPositionLast() ? 0 :
currentView.toPosition + 1;
+ }
+
+ if (index >= dirtyPages.size()) {
+ return null;
+ }
+
+ List<FullPageId> pageIds = dirtyPages.get(index).getValue();
+
+ if (fromPosition == pageIds.size() - 1 ||
!equalsByGroupAndPartition(pageIds.get(fromPosition), pageIds.get(fromPosition
+ 1))) {
+ return new CheckpointDirtyPagesView(index, fromPosition,
fromPosition);
+ }
+
+ FullPageId startPageId = pageIds.get(fromPosition);
+ FullPageId endPageId = new
FullPageId(pageId(partitionId(startPageId.pageId()) + 1, (byte) 0, 0),
startPageId.groupId());
+
+ int toPosition = binarySearch(pageIds.subList(fromPosition,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toPosition = toPosition > 0 ? toPosition - 1 : -toPosition - 2;
+
+ return new CheckpointDirtyPagesView(index, fromPosition, fromPosition
+ toPosition);
+ }
+
+ /**
+ * Queue of dirty pages that will need to be written to a checkpoint.
+ *
+ * <p>Thread safe.
+ */
+ class CheckpointDirtyPagesQueue {
+ /** Current position in the queue. */
+ private final AtomicInteger position = new AtomicInteger();
+
+ /** Sizes each element in {@link #dirtyPages} + the previous value in
this array. */
Review Comment:
```suggestion
/** Sizes of each element in {@link #dirtyPages} + the previous
value in this array. */
```
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
+ static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
+ .comparingInt(FullPageId::groupId)
+ .thenComparingLong(FullPageId::effectivePageId);
+
+ /** Empty checkpoint dirty pages. */
+ static final CheckpointDirtyPages EMPTY = new
CheckpointDirtyPages(List.of());
+
+ /** Sorted dirty pages from data regions by groupId -> partitionId ->
pageIdx. */
+ private final List<IgniteBiTuple<PersistentPageMemory, List<FullPageId>>>
dirtyPages;
+
+ /** Total number of dirty pages. */
+ private final int dirtyPagesCount;
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(Map<PersistentPageMemory, List<FullPageId>>
dirtyPages) {
+ this(dirtyPages.isEmpty() ? List.of()
+ : dirtyPages.entrySet().stream().map(e -> new
IgniteBiTuple<>(e.getKey(), e.getValue())).collect(toList()));
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(List<IgniteBiTuple<PersistentPageMemory,
List<FullPageId>>> dirtyPages) {
+ assert dirtyPages instanceof RandomAccess : dirtyPages;
+
+ this.dirtyPages = dirtyPages;
+
+ int count = 0;
+
+ for (IgniteBiTuple<PersistentPageMemory, List<FullPageId>> pages :
dirtyPages) {
+ assert !pages.getValue().isEmpty() : pages.getKey();
+ assert pages.getValue() instanceof RandomAccess : pages.getValue();
+
+ count += pages.getValue().size();
+ }
+
+ dirtyPagesCount = count;
+ }
+
+ /**
+ * Returns total number of dirty pages.
+ */
+ public int dirtyPagesCount() {
+ return dirtyPagesCount;
+ }
+
+ /**
+ * Returns a queue of dirty pages to be written to a checkpoint.
+ */
+ public CheckpointDirtyPagesQueue toQueue() {
+ return new CheckpointDirtyPagesQueue();
+ }
+
+ /**
+ * Looks for dirty page views for a specific group and partition.
+ *
+ * @param grpId Group ID.
+ * @param partId Partition ID.
+ */
+ public @Nullable CheckpointDirtyPagesView findView(int grpId, int partId) {
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ FullPageId startPageId = new FullPageId(pageId(partId, (byte) 0, 0),
grpId);
+ FullPageId endPageId = new FullPageId(pageId(partId + 1, (byte) 0, 0),
grpId);
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ List<FullPageId> pageIds = dirtyPages.get(i).getValue();
+
+ int fromIndex = binarySearch(pageIds, startPageId,
DIRTY_PAGE_COMPARATOR);
+
+ fromIndex = fromIndex >= 0 ? fromIndex : Math.min(pageIds.size() -
1, -fromIndex - 1);
+
+ if (!equalsByGroupAndPartition(startPageId,
pageIds.get(fromIndex))) {
+ continue;
+ }
+
+ int toIndex = binarySearch(pageIds.subList(fromIndex,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toIndex = toIndex > 0 ? toIndex - 1 : -toIndex - 2;
+
+ return new CheckpointDirtyPagesView(i, fromIndex, fromIndex +
toIndex);
+ }
+
+ return null;
+ }
+
+ /**
+ * Looks for the next dirty page 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 nextView(@Nullable
CheckpointDirtyPagesView currentView) {
+ assert currentView == null || currentView.owner() == this :
currentView;
+
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ int index;
+ int fromPosition;
+
+ if (currentView == null) {
+ index = 0;
+ fromPosition = 0;
+ } else {
+ index = currentView.isToPositionLast() ? currentView.index + 1 :
currentView.index;
+ fromPosition = currentView.isToPositionLast() ? 0 :
currentView.toPosition + 1;
+ }
+
+ if (index >= dirtyPages.size()) {
+ return null;
+ }
+
+ List<FullPageId> pageIds = dirtyPages.get(index).getValue();
+
+ if (fromPosition == pageIds.size() - 1 ||
!equalsByGroupAndPartition(pageIds.get(fromPosition), pageIds.get(fromPosition
+ 1))) {
+ return new CheckpointDirtyPagesView(index, fromPosition,
fromPosition);
+ }
+
+ FullPageId startPageId = pageIds.get(fromPosition);
+ FullPageId endPageId = new
FullPageId(pageId(partitionId(startPageId.pageId()) + 1, (byte) 0, 0),
startPageId.groupId());
+
+ int toPosition = binarySearch(pageIds.subList(fromPosition,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toPosition = toPosition > 0 ? toPosition - 1 : -toPosition - 2;
+
+ return new CheckpointDirtyPagesView(index, fromPosition, fromPosition
+ toPosition);
+ }
+
+ /**
+ * Queue of dirty pages that will need to be written to a checkpoint.
+ *
+ * <p>Thread safe.
+ */
+ class CheckpointDirtyPagesQueue {
+ /** Current position in the queue. */
+ private final AtomicInteger position = new AtomicInteger();
+
+ /** Sizes each element in {@link #dirtyPages} + the previous value in
this array. */
+ private final int[] sizes;
+
+ /**
+ * Private constructor.
+ */
+ private CheckpointDirtyPagesQueue() {
+ int size = 0;
+
+ int[] sizes = new int[dirtyPages.size()];
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ sizes[i] = size += dirtyPages.get(i).getValue().size();
+ }
+
+ this.sizes = sizes;
+ }
+
+ /**
+ * Returns {@link true} if the next element of the queue was obtained.
+ *
+ * @param result Holder is the result of getting the next dirty page.
+ */
+ public boolean next(QueueResult result) {
+ int queuePosition = this.position.getAndIncrement();
+
+ if (queuePosition >= dirtyPagesCount) {
+ result.owner = null;
+
+ return false;
+ }
+
+ if (result.owner != this) {
+ result.owner = this;
+ result.index = 0;
+ }
+
+ int index = result.index;
+
+ if (queuePosition >= sizes[index]) {
+ if (queuePosition == sizes[index]) {
+ index++;
+ } else {
+ index = findDirtyPagesIndex(index, queuePosition);
Review Comment:
It looks like we are already sure that `index` cannot be found by binary
search. How about adding `+1` to the first argument to skip it?
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
+ static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
+ .comparingInt(FullPageId::groupId)
+ .thenComparingLong(FullPageId::effectivePageId);
+
+ /** Empty checkpoint dirty pages. */
+ static final CheckpointDirtyPages EMPTY = new
CheckpointDirtyPages(List.of());
+
+ /** Sorted dirty pages from data regions by groupId -> partitionId ->
pageIdx. */
+ private final List<IgniteBiTuple<PersistentPageMemory, List<FullPageId>>>
dirtyPages;
+
+ /** Total number of dirty pages. */
+ private final int dirtyPagesCount;
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(Map<PersistentPageMemory, List<FullPageId>>
dirtyPages) {
+ this(dirtyPages.isEmpty() ? List.of()
+ : dirtyPages.entrySet().stream().map(e -> new
IgniteBiTuple<>(e.getKey(), e.getValue())).collect(toList()));
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(List<IgniteBiTuple<PersistentPageMemory,
List<FullPageId>>> dirtyPages) {
+ assert dirtyPages instanceof RandomAccess : dirtyPages;
+
+ this.dirtyPages = dirtyPages;
+
+ int count = 0;
+
+ for (IgniteBiTuple<PersistentPageMemory, List<FullPageId>> pages :
dirtyPages) {
+ assert !pages.getValue().isEmpty() : pages.getKey();
+ assert pages.getValue() instanceof RandomAccess : pages.getValue();
+
+ count += pages.getValue().size();
+ }
+
+ dirtyPagesCount = count;
+ }
+
+ /**
+ * Returns total number of dirty pages.
+ */
+ public int dirtyPagesCount() {
+ return dirtyPagesCount;
+ }
+
+ /**
+ * Returns a queue of dirty pages to be written to a checkpoint.
+ */
+ public CheckpointDirtyPagesQueue toQueue() {
+ return new CheckpointDirtyPagesQueue();
+ }
+
+ /**
+ * Looks for dirty page views for a specific group and partition.
+ *
+ * @param grpId Group ID.
+ * @param partId Partition ID.
+ */
+ public @Nullable CheckpointDirtyPagesView findView(int grpId, int partId) {
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ FullPageId startPageId = new FullPageId(pageId(partId, (byte) 0, 0),
grpId);
+ FullPageId endPageId = new FullPageId(pageId(partId + 1, (byte) 0, 0),
grpId);
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ List<FullPageId> pageIds = dirtyPages.get(i).getValue();
+
+ int fromIndex = binarySearch(pageIds, startPageId,
DIRTY_PAGE_COMPARATOR);
+
+ fromIndex = fromIndex >= 0 ? fromIndex : Math.min(pageIds.size() -
1, -fromIndex - 1);
+
+ if (!equalsByGroupAndPartition(startPageId,
pageIds.get(fromIndex))) {
+ continue;
+ }
+
+ int toIndex = binarySearch(pageIds.subList(fromIndex,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toIndex = toIndex > 0 ? toIndex - 1 : -toIndex - 2;
+
+ return new CheckpointDirtyPagesView(i, fromIndex, fromIndex +
toIndex);
+ }
+
+ return null;
+ }
+
+ /**
+ * Looks for the next dirty page 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 nextView(@Nullable
CheckpointDirtyPagesView currentView) {
+ assert currentView == null || currentView.owner() == this :
currentView;
+
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ int index;
+ int fromPosition;
+
+ if (currentView == null) {
+ index = 0;
+ fromPosition = 0;
+ } else {
+ index = currentView.isToPositionLast() ? currentView.index + 1 :
currentView.index;
+ fromPosition = currentView.isToPositionLast() ? 0 :
currentView.toPosition + 1;
+ }
+
+ if (index >= dirtyPages.size()) {
+ return null;
+ }
+
+ List<FullPageId> pageIds = dirtyPages.get(index).getValue();
+
+ if (fromPosition == pageIds.size() - 1 ||
!equalsByGroupAndPartition(pageIds.get(fromPosition), pageIds.get(fromPosition
+ 1))) {
+ return new CheckpointDirtyPagesView(index, fromPosition,
fromPosition);
+ }
+
+ FullPageId startPageId = pageIds.get(fromPosition);
+ FullPageId endPageId = new
FullPageId(pageId(partitionId(startPageId.pageId()) + 1, (byte) 0, 0),
startPageId.groupId());
+
+ int toPosition = binarySearch(pageIds.subList(fromPosition,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toPosition = toPosition > 0 ? toPosition - 1 : -toPosition - 2;
+
+ return new CheckpointDirtyPagesView(index, fromPosition, fromPosition
+ toPosition);
+ }
+
+ /**
+ * Queue of dirty pages that will need to be written to a checkpoint.
+ *
+ * <p>Thread safe.
+ */
+ class CheckpointDirtyPagesQueue {
+ /** Current position in the queue. */
+ private final AtomicInteger position = new AtomicInteger();
+
+ /** Sizes each element in {@link #dirtyPages} + the previous value in
this array. */
+ private final int[] sizes;
+
+ /**
+ * Private constructor.
+ */
+ private CheckpointDirtyPagesQueue() {
+ int size = 0;
+
+ int[] sizes = new int[dirtyPages.size()];
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ sizes[i] = size += dirtyPages.get(i).getValue().size();
+ }
+
+ this.sizes = sizes;
+ }
+
+ /**
+ * Returns {@link true} if the next element of the queue was obtained.
+ *
+ * @param result Holder is the result of getting the next dirty page.
+ */
+ public boolean next(QueueResult result) {
+ int queuePosition = this.position.getAndIncrement();
+
+ if (queuePosition >= dirtyPagesCount) {
+ result.owner = null;
+
+ return false;
+ }
+
+ if (result.owner != this) {
+ result.owner = this;
+ result.index = 0;
+ }
+
+ int index = result.index;
+
+ if (queuePosition >= sizes[index]) {
+ if (queuePosition == sizes[index]) {
+ index++;
+ } else {
+ index = findDirtyPagesIndex(index, queuePosition);
+ }
+ }
+
+ result.index = index;
+ result.position = index > 0 ? queuePosition - sizes[index - 1] :
queuePosition;
+
+ return true;
+ }
+
+ /**
+ * Returns {@link true} if the queue is empty.
+ */
+ public boolean isEmpty() {
+ return position.get() >= dirtyPagesCount;
+ }
+
+ /**
+ * Returns the size of the queue.
+ */
+ public int size() {
+ return dirtyPagesCount - Math.min(dirtyPagesCount, position.get());
+ }
+
+ private int findDirtyPagesIndex(int index, int position) {
+ return Math.abs(Arrays.binarySearch(sizes, index, sizes.length,
position) + 1);
+ }
+
+ private CheckpointDirtyPages owner() {
+ return CheckpointDirtyPages.this;
+ }
+ }
+
+ /**
+ * View of {@link CheckpointDirtyPages} in which all dirty pages will
refer to the same {@link PersistentPageMemory} and contain the
+ * same groupId and partitionId and increasing pageIdx.
+ *
+ * <p>Thread safe.
+ */
+ class CheckpointDirtyPagesView {
+ /** Element index in {@link CheckpointDirtyPages#dirtyPages}. */
+ private final int index;
+
+ /** Starting position (inclusive) of the dirty page within the element
at {@link #index}. */
+ private final int fromPosition;
+
+ /** End position (inclusive) of the dirty page within the element at
{@link #index}. */
+ private final int toPosition;
+
+ /**
+ * Private constructor.
+ *
+ * @param index Element index in {@link
CheckpointDirtyPages#dirtyPages}.
+ * @param fromPosition Starting position (inclusive) of the dirty page
within the element at {@link #index}.
+ * @param toPosition End position (inclusive) of the dirty page within
the element at {@link #index}.
+ */
+ private CheckpointDirtyPagesView(int index, int fromPosition, int
toPosition) {
+ this.index = index;
+ this.fromPosition = fromPosition;
+ this.toPosition = toPosition;
+ }
+
+ /**
+ * Returns the dirty page by index.
+ *
+ * @param index Dirty page index.
+ */
+ public FullPageId get(int index) {
+ return dirtyPages.get(this.index).getValue().get(fromPosition +
index);
+ }
+
+ /**
+ * Returns the page memory for view.
+ */
+ public PersistentPageMemory pageMemory() {
+ return dirtyPages.get(index).getKey();
+ }
+
+ /**
+ * Returns the size of the view.
+ */
+ public int size() {
+ return toPosition - fromPosition + 1;
+ }
+
+ private CheckpointDirtyPages owner() {
+ return CheckpointDirtyPages.this;
+ }
+
+ private boolean isToPositionLast() {
+ return toPosition == dirtyPages.get(index).getValue().size() - 1;
+ }
+ }
+
+ /**
+ * Holder is the result of getting the next dirty page in {@link
CheckpointDirtyPagesQueue#next(QueueResult)}.
+ *
+ * <p>Not thread safe.
+ */
+ static class QueueResult {
+ private @Nullable CheckpointDirtyPagesQueue owner;
+
+ /** Element index in {@link CheckpointDirtyPages#dirtyPages}. */
+ private int index;
+
+ /** Position of the dirty page within the element at {@link #index}. */
+ private int position;
+
+ /**
+ * Returns the page memory for the associated dirty page.
+ */
+ public @Nullable PersistentPageMemory pageMemory() {
+ return owner == null ? null :
owner.owner().dirtyPages.get(index).getKey();
+ }
+
+ /**
+ * Returns dirty page.
+ */
+ public @Nullable FullPageId dirtyPage() {
+ return owner == null ? null :
owner.owner().dirtyPages.get(index).getValue().get(position);
+ }
+ }
+
+ private static boolean equalsByGroupAndPartition(FullPageId pageId0,
FullPageId pageId1) {
Review Comment:
We usually put all member types last in a file (after all methods). Is this
a rule in Ignite 3 code? If yes, this method should probably be moved before
the member classes.
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/Checkpointer.java:
##########
@@ -379,10 +378,12 @@ boolean writePages(
tracker.onPagesWriteStart();
+ CheckpointDirtyPagesQueue checkpointDirtyPagesQueue =
checkpointDirtyPages.toQueue();
Review Comment:
Should we just inline it if it's just used in one place?
##########
modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPagesTest.java:
##########
@@ -0,0 +1,287 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.pagememory.persistence.checkpoint;
+
+import static java.util.stream.Collectors.toList;
+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.util.PageIdUtils.partitionId;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.runAsync;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+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.concurrent.TimeUnit;
+import java.util.function.Predicate;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import
org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.CheckpointDirtyPagesQueue;
+import
org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.CheckpointDirtyPagesView;
+import
org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.QueueResult;
+import org.apache.ignite.internal.pagememory.util.PageIdUtils;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.junit.jupiter.api.Test;
+
+/**
+ * For {@link CheckpointDirtyPages} testing.
+ */
+public class CheckpointDirtyPagesTest {
+ @Test
+ void testDirtyPagesCount() {
+ var dirtyPages0 = createDirtyPages(of(0, 0, 0), of(0, 0, 1));
Review Comment:
I think the convention is to only use `var` with `new` operator or with an
explicit cast. Should an explicit type be used instead?
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
+ static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
+ .comparingInt(FullPageId::groupId)
+ .thenComparingLong(FullPageId::effectivePageId);
+
+ /** Empty checkpoint dirty pages. */
+ static final CheckpointDirtyPages EMPTY = new
CheckpointDirtyPages(List.of());
+
+ /** Sorted dirty pages from data regions by groupId -> partitionId ->
pageIdx. */
+ private final List<IgniteBiTuple<PersistentPageMemory, List<FullPageId>>>
dirtyPages;
+
+ /** Total number of dirty pages. */
+ private final int dirtyPagesCount;
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(Map<PersistentPageMemory, List<FullPageId>>
dirtyPages) {
+ this(dirtyPages.isEmpty() ? List.of()
+ : dirtyPages.entrySet().stream().map(e -> new
IgniteBiTuple<>(e.getKey(), e.getValue())).collect(toList()));
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(List<IgniteBiTuple<PersistentPageMemory,
List<FullPageId>>> dirtyPages) {
+ assert dirtyPages instanceof RandomAccess : dirtyPages;
+
+ this.dirtyPages = dirtyPages;
+
+ int count = 0;
+
+ for (IgniteBiTuple<PersistentPageMemory, List<FullPageId>> pages :
dirtyPages) {
+ assert !pages.getValue().isEmpty() : pages.getKey();
+ assert pages.getValue() instanceof RandomAccess : pages.getValue();
+
+ count += pages.getValue().size();
+ }
+
+ dirtyPagesCount = count;
+ }
+
+ /**
+ * Returns total number of dirty pages.
+ */
+ public int dirtyPagesCount() {
+ return dirtyPagesCount;
+ }
+
+ /**
+ * Returns a queue of dirty pages to be written to a checkpoint.
+ */
+ public CheckpointDirtyPagesQueue toQueue() {
+ return new CheckpointDirtyPagesQueue();
+ }
+
+ /**
+ * Looks for dirty page views for a specific group and partition.
+ *
+ * @param grpId Group ID.
+ * @param partId Partition ID.
+ */
+ public @Nullable CheckpointDirtyPagesView findView(int grpId, int partId) {
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ FullPageId startPageId = new FullPageId(pageId(partId, (byte) 0, 0),
grpId);
+ FullPageId endPageId = new FullPageId(pageId(partId + 1, (byte) 0, 0),
grpId);
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ List<FullPageId> pageIds = dirtyPages.get(i).getValue();
+
+ int fromIndex = binarySearch(pageIds, startPageId,
DIRTY_PAGE_COMPARATOR);
+
+ fromIndex = fromIndex >= 0 ? fromIndex : Math.min(pageIds.size() -
1, -fromIndex - 1);
+
+ if (!equalsByGroupAndPartition(startPageId,
pageIds.get(fromIndex))) {
+ continue;
+ }
+
+ int toIndex = binarySearch(pageIds.subList(fromIndex,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toIndex = toIndex > 0 ? toIndex - 1 : -toIndex - 2;
+
+ return new CheckpointDirtyPagesView(i, fromIndex, fromIndex +
toIndex);
+ }
+
+ return null;
+ }
+
+ /**
+ * Looks for the next dirty page 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 nextView(@Nullable
CheckpointDirtyPagesView currentView) {
+ assert currentView == null || currentView.owner() == this :
currentView;
+
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ int index;
+ int fromPosition;
+
+ if (currentView == null) {
+ index = 0;
+ fromPosition = 0;
+ } else {
+ index = currentView.isToPositionLast() ? currentView.index + 1 :
currentView.index;
+ fromPosition = currentView.isToPositionLast() ? 0 :
currentView.toPosition + 1;
+ }
+
+ if (index >= dirtyPages.size()) {
+ return null;
+ }
+
+ List<FullPageId> pageIds = dirtyPages.get(index).getValue();
+
+ if (fromPosition == pageIds.size() - 1 ||
!equalsByGroupAndPartition(pageIds.get(fromPosition), pageIds.get(fromPosition
+ 1))) {
+ return new CheckpointDirtyPagesView(index, fromPosition,
fromPosition);
+ }
+
+ FullPageId startPageId = pageIds.get(fromPosition);
+ FullPageId endPageId = new
FullPageId(pageId(partitionId(startPageId.pageId()) + 1, (byte) 0, 0),
startPageId.groupId());
+
+ int toPosition = binarySearch(pageIds.subList(fromPosition,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toPosition = toPosition > 0 ? toPosition - 1 : -toPosition - 2;
+
+ return new CheckpointDirtyPagesView(index, fromPosition, fromPosition
+ toPosition);
+ }
+
+ /**
+ * Queue of dirty pages that will need to be written to a checkpoint.
+ *
+ * <p>Thread safe.
+ */
+ class CheckpointDirtyPagesQueue {
+ /** Current position in the queue. */
+ private final AtomicInteger position = new AtomicInteger();
+
+ /** Sizes each element in {@link #dirtyPages} + the previous value in
this array. */
+ private final int[] sizes;
+
+ /**
+ * Private constructor.
+ */
+ private CheckpointDirtyPagesQueue() {
+ int size = 0;
+
+ int[] sizes = new int[dirtyPages.size()];
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ sizes[i] = size += dirtyPages.get(i).getValue().size();
+ }
+
+ this.sizes = sizes;
+ }
+
+ /**
+ * Returns {@link true} if the next element of the queue was obtained.
+ *
+ * @param result Holder is the result of getting the next dirty page.
+ */
+ public boolean next(QueueResult result) {
+ int queuePosition = this.position.getAndIncrement();
+
+ if (queuePosition >= dirtyPagesCount) {
+ result.owner = null;
+
+ return false;
+ }
+
+ if (result.owner != this) {
+ result.owner = this;
+ result.index = 0;
+ }
+
+ int index = result.index;
+
+ if (queuePosition >= sizes[index]) {
+ if (queuePosition == sizes[index]) {
+ index++;
+ } else {
+ index = findDirtyPagesIndex(index, queuePosition);
+ }
+ }
+
+ result.index = index;
+ result.position = index > 0 ? queuePosition - sizes[index - 1] :
queuePosition;
+
+ return true;
+ }
+
+ /**
+ * Returns {@link true} if the queue is empty.
+ */
+ public boolean isEmpty() {
+ return position.get() >= dirtyPagesCount;
+ }
+
+ /**
+ * Returns the size of the queue.
+ */
+ public int size() {
+ return dirtyPagesCount - Math.min(dirtyPagesCount, position.get());
+ }
+
+ private int findDirtyPagesIndex(int index, int position) {
+ return Math.abs(Arrays.binarySearch(sizes, index, sizes.length,
position) + 1);
+ }
+
+ private CheckpointDirtyPages owner() {
+ return CheckpointDirtyPages.this;
+ }
+ }
+
+ /**
+ * View of {@link CheckpointDirtyPages} in which all dirty pages will
refer to the same {@link PersistentPageMemory} and contain the
+ * same groupId and partitionId and increasing pageIdx.
+ *
+ * <p>Thread safe.
+ */
+ class CheckpointDirtyPagesView {
+ /** Element index in {@link CheckpointDirtyPages#dirtyPages}. */
+ private final int index;
+
+ /** Starting position (inclusive) of the dirty page within the element
at {@link #index}. */
+ private final int fromPosition;
+
+ /** End position (inclusive) of the dirty page within the element at
{@link #index}. */
+ private final int toPosition;
+
+ /**
+ * Private constructor.
+ *
+ * @param index Element index in {@link
CheckpointDirtyPages#dirtyPages}.
+ * @param fromPosition Starting position (inclusive) of the dirty page
within the element at {@link #index}.
+ * @param toPosition End position (inclusive) of the dirty page within
the element at {@link #index}.
+ */
+ private CheckpointDirtyPagesView(int index, int fromPosition, int
toPosition) {
+ this.index = index;
+ this.fromPosition = fromPosition;
+ this.toPosition = toPosition;
+ }
+
+ /**
+ * Returns the dirty page by index.
+ *
+ * @param index Dirty page index.
+ */
+ public FullPageId get(int index) {
Review Comment:
Both the parameter and a field have name `index`, but they mean different
things. How about renaming the field (not the parameter) to something more
descriptive, like `regionIndex` or `pageMemoryIndex`, to avoid the clash?
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
Review Comment:
It looks more like a dirty page ID comparator, not page comparator
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriter.java:
##########
@@ -51,8 +53,8 @@ public class CheckpointPagesWriter implements Runnable {
/** Checkpoint specific metrics 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 final IgniteConcurrentMultiPairQueue<PersistentPageMemory,
FullPageId> writePageIds;
+ /** Queue of dirty page IDs to write under this task. Overall pages to
write may be greater than this queue. */
Review Comment:
When will we write more than the IDs contained in this queue? Could you
please clarify this in this javadoc too?
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
+ static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
+ .comparingInt(FullPageId::groupId)
+ .thenComparingLong(FullPageId::effectivePageId);
+
+ /** Empty checkpoint dirty pages. */
+ static final CheckpointDirtyPages EMPTY = new
CheckpointDirtyPages(List.of());
+
+ /** Sorted dirty pages from data regions by groupId -> partitionId ->
pageIdx. */
+ private final List<IgniteBiTuple<PersistentPageMemory, List<FullPageId>>>
dirtyPages;
+
+ /** Total number of dirty pages. */
+ private final int dirtyPagesCount;
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(Map<PersistentPageMemory, List<FullPageId>>
dirtyPages) {
+ this(dirtyPages.isEmpty() ? List.of()
+ : dirtyPages.entrySet().stream().map(e -> new
IgniteBiTuple<>(e.getKey(), e.getValue())).collect(toList()));
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(List<IgniteBiTuple<PersistentPageMemory,
List<FullPageId>>> dirtyPages) {
+ assert dirtyPages instanceof RandomAccess : dirtyPages;
+
+ this.dirtyPages = dirtyPages;
+
+ int count = 0;
+
+ for (IgniteBiTuple<PersistentPageMemory, List<FullPageId>> pages :
dirtyPages) {
+ assert !pages.getValue().isEmpty() : pages.getKey();
+ assert pages.getValue() instanceof RandomAccess : pages.getValue();
+
+ count += pages.getValue().size();
+ }
+
+ dirtyPagesCount = count;
+ }
+
+ /**
+ * Returns total number of dirty pages.
+ */
+ public int dirtyPagesCount() {
+ return dirtyPagesCount;
+ }
+
+ /**
+ * Returns a queue of dirty pages to be written to a checkpoint.
+ */
+ public CheckpointDirtyPagesQueue toQueue() {
+ return new CheckpointDirtyPagesQueue();
+ }
+
+ /**
+ * Looks for dirty page views for a specific group and partition.
+ *
+ * @param grpId Group ID.
+ * @param partId Partition ID.
+ */
+ public @Nullable CheckpointDirtyPagesView findView(int grpId, int partId) {
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ FullPageId startPageId = new FullPageId(pageId(partId, (byte) 0, 0),
grpId);
+ FullPageId endPageId = new FullPageId(pageId(partId + 1, (byte) 0, 0),
grpId);
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ List<FullPageId> pageIds = dirtyPages.get(i).getValue();
+
+ int fromIndex = binarySearch(pageIds, startPageId,
DIRTY_PAGE_COMPARATOR);
+
+ fromIndex = fromIndex >= 0 ? fromIndex : Math.min(pageIds.size() -
1, -fromIndex - 1);
+
+ if (!equalsByGroupAndPartition(startPageId,
pageIds.get(fromIndex))) {
+ continue;
+ }
+
+ int toIndex = binarySearch(pageIds.subList(fromIndex,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toIndex = toIndex > 0 ? toIndex - 1 : -toIndex - 2;
+
+ return new CheckpointDirtyPagesView(i, fromIndex, fromIndex +
toIndex);
+ }
+
+ return null;
+ }
+
+ /**
+ * Looks for the next dirty page 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 nextView(@Nullable
CheckpointDirtyPagesView currentView) {
+ assert currentView == null || currentView.owner() == this :
currentView;
+
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ int index;
Review Comment:
`index` -> `pageMemoryIndex` ? Because there are many 'indexable' things in
the context, it's easy to get lost.
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
+ static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
+ .comparingInt(FullPageId::groupId)
+ .thenComparingLong(FullPageId::effectivePageId);
+
+ /** Empty checkpoint dirty pages. */
+ static final CheckpointDirtyPages EMPTY = new
CheckpointDirtyPages(List.of());
+
+ /** Sorted dirty pages from data regions by groupId -> partitionId ->
pageIdx. */
+ private final List<IgniteBiTuple<PersistentPageMemory, List<FullPageId>>>
dirtyPages;
+
+ /** Total number of dirty pages. */
+ private final int dirtyPagesCount;
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(Map<PersistentPageMemory, List<FullPageId>>
dirtyPages) {
+ this(dirtyPages.isEmpty() ? List.of()
+ : dirtyPages.entrySet().stream().map(e -> new
IgniteBiTuple<>(e.getKey(), e.getValue())).collect(toList()));
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(List<IgniteBiTuple<PersistentPageMemory,
List<FullPageId>>> dirtyPages) {
+ assert dirtyPages instanceof RandomAccess : dirtyPages;
+
+ this.dirtyPages = dirtyPages;
+
+ int count = 0;
+
+ for (IgniteBiTuple<PersistentPageMemory, List<FullPageId>> pages :
dirtyPages) {
+ assert !pages.getValue().isEmpty() : pages.getKey();
+ assert pages.getValue() instanceof RandomAccess : pages.getValue();
+
+ count += pages.getValue().size();
+ }
+
+ dirtyPagesCount = count;
+ }
+
+ /**
+ * Returns total number of dirty pages.
+ */
+ public int dirtyPagesCount() {
+ return dirtyPagesCount;
+ }
+
+ /**
+ * Returns a queue of dirty pages to be written to a checkpoint.
+ */
+ public CheckpointDirtyPagesQueue toQueue() {
+ return new CheckpointDirtyPagesQueue();
+ }
+
+ /**
+ * Looks for dirty page views for a specific group and partition.
+ *
+ * @param grpId Group ID.
+ * @param partId Partition ID.
+ */
+ public @Nullable CheckpointDirtyPagesView findView(int grpId, int partId) {
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ FullPageId startPageId = new FullPageId(pageId(partId, (byte) 0, 0),
grpId);
+ FullPageId endPageId = new FullPageId(pageId(partId + 1, (byte) 0, 0),
grpId);
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ List<FullPageId> pageIds = dirtyPages.get(i).getValue();
+
+ int fromIndex = binarySearch(pageIds, startPageId,
DIRTY_PAGE_COMPARATOR);
+
+ fromIndex = fromIndex >= 0 ? fromIndex : Math.min(pageIds.size() -
1, -fromIndex - 1);
+
+ if (!equalsByGroupAndPartition(startPageId,
pageIds.get(fromIndex))) {
+ continue;
+ }
+
+ int toIndex = binarySearch(pageIds.subList(fromIndex,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toIndex = toIndex > 0 ? toIndex - 1 : -toIndex - 2;
+
+ return new CheckpointDirtyPagesView(i, fromIndex, fromIndex +
toIndex);
+ }
+
+ return null;
+ }
+
+ /**
+ * Looks for the next dirty page 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 nextView(@Nullable
CheckpointDirtyPagesView currentView) {
+ assert currentView == null || currentView.owner() == this :
currentView;
+
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ int index;
+ int fromPosition;
+
+ if (currentView == null) {
+ index = 0;
+ fromPosition = 0;
+ } else {
+ index = currentView.isToPositionLast() ? currentView.index + 1 :
currentView.index;
+ fromPosition = currentView.isToPositionLast() ? 0 :
currentView.toPosition + 1;
+ }
+
+ if (index >= dirtyPages.size()) {
+ return null;
+ }
+
+ List<FullPageId> pageIds = dirtyPages.get(index).getValue();
+
+ if (fromPosition == pageIds.size() - 1 ||
!equalsByGroupAndPartition(pageIds.get(fromPosition), pageIds.get(fromPosition
+ 1))) {
+ return new CheckpointDirtyPagesView(index, fromPosition,
fromPosition);
+ }
+
+ FullPageId startPageId = pageIds.get(fromPosition);
+ FullPageId endPageId = new
FullPageId(pageId(partitionId(startPageId.pageId()) + 1, (byte) 0, 0),
startPageId.groupId());
+
+ int toPosition = binarySearch(pageIds.subList(fromPosition,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toPosition = toPosition > 0 ? toPosition - 1 : -toPosition - 2;
+
+ return new CheckpointDirtyPagesView(index, fromPosition, fromPosition
+ toPosition);
+ }
+
+ /**
+ * Queue of dirty pages that will need to be written to a checkpoint.
+ *
+ * <p>Thread safe.
+ */
+ class CheckpointDirtyPagesQueue {
+ /** Current position in the queue. */
+ private final AtomicInteger position = new AtomicInteger();
+
+ /** Sizes each element in {@link #dirtyPages} + the previous value in
this array. */
+ private final int[] sizes;
Review Comment:
This array contains cumulative sizes, so I suggest renaming it (like
`cumulativeSizes`) to make it obvious
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
+ static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
+ .comparingInt(FullPageId::groupId)
+ .thenComparingLong(FullPageId::effectivePageId);
+
+ /** Empty checkpoint dirty pages. */
+ static final CheckpointDirtyPages EMPTY = new
CheckpointDirtyPages(List.of());
+
+ /** Sorted dirty pages from data regions by groupId -> partitionId ->
pageIdx. */
+ private final List<IgniteBiTuple<PersistentPageMemory, List<FullPageId>>>
dirtyPages;
+
+ /** Total number of dirty pages. */
+ private final int dirtyPagesCount;
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(Map<PersistentPageMemory, List<FullPageId>>
dirtyPages) {
+ this(dirtyPages.isEmpty() ? List.of()
+ : dirtyPages.entrySet().stream().map(e -> new
IgniteBiTuple<>(e.getKey(), e.getValue())).collect(toList()));
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(List<IgniteBiTuple<PersistentPageMemory,
List<FullPageId>>> dirtyPages) {
+ assert dirtyPages instanceof RandomAccess : dirtyPages;
+
+ this.dirtyPages = dirtyPages;
+
+ int count = 0;
+
+ for (IgniteBiTuple<PersistentPageMemory, List<FullPageId>> pages :
dirtyPages) {
+ assert !pages.getValue().isEmpty() : pages.getKey();
+ assert pages.getValue() instanceof RandomAccess : pages.getValue();
+
+ count += pages.getValue().size();
+ }
+
+ dirtyPagesCount = count;
+ }
+
+ /**
+ * Returns total number of dirty pages.
+ */
+ public int dirtyPagesCount() {
+ return dirtyPagesCount;
+ }
+
+ /**
+ * Returns a queue of dirty pages to be written to a checkpoint.
+ */
+ public CheckpointDirtyPagesQueue toQueue() {
+ return new CheckpointDirtyPagesQueue();
+ }
+
+ /**
+ * Looks for dirty page views for a specific group and partition.
+ *
+ * @param grpId Group ID.
+ * @param partId Partition ID.
+ */
+ public @Nullable CheckpointDirtyPagesView findView(int grpId, int partId) {
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ FullPageId startPageId = new FullPageId(pageId(partId, (byte) 0, 0),
grpId);
+ FullPageId endPageId = new FullPageId(pageId(partId + 1, (byte) 0, 0),
grpId);
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ List<FullPageId> pageIds = dirtyPages.get(i).getValue();
+
+ int fromIndex = binarySearch(pageIds, startPageId,
DIRTY_PAGE_COMPARATOR);
+
+ fromIndex = fromIndex >= 0 ? fromIndex : Math.min(pageIds.size() -
1, -fromIndex - 1);
+
+ if (!equalsByGroupAndPartition(startPageId,
pageIds.get(fromIndex))) {
+ continue;
+ }
+
+ int toIndex = binarySearch(pageIds.subList(fromIndex,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toIndex = toIndex > 0 ? toIndex - 1 : -toIndex - 2;
+
+ return new CheckpointDirtyPagesView(i, fromIndex, fromIndex +
toIndex);
+ }
+
+ return null;
+ }
+
+ /**
+ * Looks for the next dirty page 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 nextView(@Nullable
CheckpointDirtyPagesView currentView) {
+ assert currentView == null || currentView.owner() == this :
currentView;
+
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ int index;
+ int fromPosition;
+
+ if (currentView == null) {
+ index = 0;
+ fromPosition = 0;
+ } else {
+ index = currentView.isToPositionLast() ? currentView.index + 1 :
currentView.index;
+ fromPosition = currentView.isToPositionLast() ? 0 :
currentView.toPosition + 1;
+ }
+
+ if (index >= dirtyPages.size()) {
+ return null;
+ }
+
+ List<FullPageId> pageIds = dirtyPages.get(index).getValue();
+
+ if (fromPosition == pageIds.size() - 1 ||
!equalsByGroupAndPartition(pageIds.get(fromPosition), pageIds.get(fromPosition
+ 1))) {
+ return new CheckpointDirtyPagesView(index, fromPosition,
fromPosition);
+ }
+
+ FullPageId startPageId = pageIds.get(fromPosition);
+ FullPageId endPageId = new
FullPageId(pageId(partitionId(startPageId.pageId()) + 1, (byte) 0, 0),
startPageId.groupId());
+
+ int toPosition = binarySearch(pageIds.subList(fromPosition,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toPosition = toPosition > 0 ? toPosition - 1 : -toPosition - 2;
+
+ return new CheckpointDirtyPagesView(index, fromPosition, fromPosition
+ toPosition);
+ }
+
+ /**
+ * Queue of dirty pages that will need to be written to a checkpoint.
+ *
+ * <p>Thread safe.
+ */
+ class CheckpointDirtyPagesQueue {
+ /** Current position in the queue. */
+ private final AtomicInteger position = new AtomicInteger();
+
+ /** Sizes each element in {@link #dirtyPages} + the previous value in
this array. */
+ private final int[] sizes;
+
+ /**
+ * Private constructor.
+ */
+ private CheckpointDirtyPagesQueue() {
+ int size = 0;
+
+ int[] sizes = new int[dirtyPages.size()];
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ sizes[i] = size += dirtyPages.get(i).getValue().size();
+ }
+
+ this.sizes = sizes;
+ }
+
+ /**
+ * Returns {@link true} if the next element of the queue was obtained.
+ *
+ * @param result Holder is the result of getting the next dirty page.
+ */
+ public boolean next(QueueResult result) {
+ int queuePosition = this.position.getAndIncrement();
+
+ if (queuePosition >= dirtyPagesCount) {
+ result.owner = null;
+
+ return false;
+ }
+
+ if (result.owner != this) {
+ result.owner = this;
+ result.index = 0;
+ }
+
+ int index = result.index;
+
+ if (queuePosition >= sizes[index]) {
+ if (queuePosition == sizes[index]) {
+ index++;
+ } else {
+ index = findDirtyPagesIndex(index, queuePosition);
+ }
+ }
+
+ result.index = index;
+ result.position = index > 0 ? queuePosition - sizes[index - 1] :
queuePosition;
+
+ return true;
+ }
+
+ /**
+ * Returns {@link true} if the queue is empty.
+ */
+ public boolean isEmpty() {
+ return position.get() >= dirtyPagesCount;
+ }
+
+ /**
+ * Returns the size of the queue.
+ */
+ public int size() {
+ return dirtyPagesCount - Math.min(dirtyPagesCount, position.get());
+ }
+
+ private int findDirtyPagesIndex(int index, int position) {
+ return Math.abs(Arrays.binarySearch(sizes, index, sizes.length,
position) + 1);
Review Comment:
Can a negative value be returned here at all? As, if the position is more
than the whole sum of all sizes, we just should not be here.
If a negative is not a possibility, let's remove `abs()` and just assert
that the value is non-negative.
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointWorkflow.java:
##########
@@ -63,14 +61,8 @@
* <p>{@link CheckpointWorkflow#markCheckpointEnd} - Finalization of last
checkpoint.
*/
class CheckpointWorkflow {
- /**
- * Starting from this number of dirty pages in checkpoint, array will be
sorted with {@link Arrays#parallelSort(Comparable[])} in case
- * of {@link CheckpointWriteOrder#SEQUENTIAL}.
- */
- private final int parallelSortThreshold;
-
- /** This number of threads will be created and used for parallel sorting.
*/
- private static final int PARALLEL_SORT_THREADS =
Math.min(Runtime.getRuntime().availableProcessors(), 8);
+ /** Starting from this number of dirty pages in checkpoint, array will be
sorted with {@link Arrays#parallelSort(Comparable[])}. */
+ static final int PARALLEL_SORT_THRESHOLD = 40_000;
Review Comment:
How about adding a link to the paper which was used to derive this value?
Otherwise, it would look like an arbitrary choice.
##########
modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointWorkflowTest.java:
##########
@@ -181,13 +171,13 @@ void testMarkCheckpointBegin() throws Exception {
CheckpointReadWriteLock readWriteLock = newReadWriteLock(log);
- List<FullPageId> dirtyPages = List.of(new FullPageId(0, 0), new
FullPageId(1, 0), new FullPageId(2, 0));
+ List<FullPageId> dirtyPages = List.of(of(0, 0, 0), of(0, 0, 1), of(0,
0, 2));
Review Comment:
`of()` method name does not seem the best choice as it gives the reader no
clue about what kind of objects it constructs. Could we rename it?
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointDirtyPages.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.pagememory.persistence.checkpoint;
+
+import static java.util.Collections.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.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.pagememory.FullPageId;
+import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sorted dirty pages from data regions that should be checkpointed.
+ *
+ * <p>Dirty pages should be sorted by groupId -> partitionId -> pageIdx.
+ */
+class CheckpointDirtyPages {
+ /** Dirty page comparator. */
+ static final Comparator<FullPageId> DIRTY_PAGE_COMPARATOR = Comparator
+ .comparingInt(FullPageId::groupId)
+ .thenComparingLong(FullPageId::effectivePageId);
+
+ /** Empty checkpoint dirty pages. */
+ static final CheckpointDirtyPages EMPTY = new
CheckpointDirtyPages(List.of());
+
+ /** Sorted dirty pages from data regions by groupId -> partitionId ->
pageIdx. */
+ private final List<IgniteBiTuple<PersistentPageMemory, List<FullPageId>>>
dirtyPages;
+
+ /** Total number of dirty pages. */
+ private final int dirtyPagesCount;
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(Map<PersistentPageMemory, List<FullPageId>>
dirtyPages) {
+ this(dirtyPages.isEmpty() ? List.of()
+ : dirtyPages.entrySet().stream().map(e -> new
IgniteBiTuple<>(e.getKey(), e.getValue())).collect(toList()));
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param dirtyPages Sorted dirty pages from data regions by groupId ->
partitionId -> pageIdx.
+ */
+ public CheckpointDirtyPages(List<IgniteBiTuple<PersistentPageMemory,
List<FullPageId>>> dirtyPages) {
+ assert dirtyPages instanceof RandomAccess : dirtyPages;
+
+ this.dirtyPages = dirtyPages;
+
+ int count = 0;
+
+ for (IgniteBiTuple<PersistentPageMemory, List<FullPageId>> pages :
dirtyPages) {
+ assert !pages.getValue().isEmpty() : pages.getKey();
+ assert pages.getValue() instanceof RandomAccess : pages.getValue();
+
+ count += pages.getValue().size();
+ }
+
+ dirtyPagesCount = count;
+ }
+
+ /**
+ * Returns total number of dirty pages.
+ */
+ public int dirtyPagesCount() {
+ return dirtyPagesCount;
+ }
+
+ /**
+ * Returns a queue of dirty pages to be written to a checkpoint.
+ */
+ public CheckpointDirtyPagesQueue toQueue() {
+ return new CheckpointDirtyPagesQueue();
+ }
+
+ /**
+ * Looks for dirty page views for a specific group and partition.
+ *
+ * @param grpId Group ID.
+ * @param partId Partition ID.
+ */
+ public @Nullable CheckpointDirtyPagesView findView(int grpId, int partId) {
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ FullPageId startPageId = new FullPageId(pageId(partId, (byte) 0, 0),
grpId);
+ FullPageId endPageId = new FullPageId(pageId(partId + 1, (byte) 0, 0),
grpId);
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ List<FullPageId> pageIds = dirtyPages.get(i).getValue();
+
+ int fromIndex = binarySearch(pageIds, startPageId,
DIRTY_PAGE_COMPARATOR);
+
+ fromIndex = fromIndex >= 0 ? fromIndex : Math.min(pageIds.size() -
1, -fromIndex - 1);
+
+ if (!equalsByGroupAndPartition(startPageId,
pageIds.get(fromIndex))) {
+ continue;
+ }
+
+ int toIndex = binarySearch(pageIds.subList(fromIndex,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toIndex = toIndex > 0 ? toIndex - 1 : -toIndex - 2;
+
+ return new CheckpointDirtyPagesView(i, fromIndex, fromIndex +
toIndex);
+ }
+
+ return null;
+ }
+
+ /**
+ * Looks for the next dirty page 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 nextView(@Nullable
CheckpointDirtyPagesView currentView) {
+ assert currentView == null || currentView.owner() == this :
currentView;
+
+ if (dirtyPages.isEmpty()) {
+ return null;
+ }
+
+ int index;
+ int fromPosition;
+
+ if (currentView == null) {
+ index = 0;
+ fromPosition = 0;
+ } else {
+ index = currentView.isToPositionLast() ? currentView.index + 1 :
currentView.index;
+ fromPosition = currentView.isToPositionLast() ? 0 :
currentView.toPosition + 1;
+ }
+
+ if (index >= dirtyPages.size()) {
+ return null;
+ }
+
+ List<FullPageId> pageIds = dirtyPages.get(index).getValue();
+
+ if (fromPosition == pageIds.size() - 1 ||
!equalsByGroupAndPartition(pageIds.get(fromPosition), pageIds.get(fromPosition
+ 1))) {
+ return new CheckpointDirtyPagesView(index, fromPosition,
fromPosition);
+ }
+
+ FullPageId startPageId = pageIds.get(fromPosition);
+ FullPageId endPageId = new
FullPageId(pageId(partitionId(startPageId.pageId()) + 1, (byte) 0, 0),
startPageId.groupId());
+
+ int toPosition = binarySearch(pageIds.subList(fromPosition,
pageIds.size()), endPageId, DIRTY_PAGE_COMPARATOR);
+
+ toPosition = toPosition > 0 ? toPosition - 1 : -toPosition - 2;
+
+ return new CheckpointDirtyPagesView(index, fromPosition, fromPosition
+ toPosition);
+ }
+
+ /**
+ * Queue of dirty pages that will need to be written to a checkpoint.
+ *
+ * <p>Thread safe.
+ */
+ class CheckpointDirtyPagesQueue {
+ /** Current position in the queue. */
+ private final AtomicInteger position = new AtomicInteger();
+
+ /** Sizes each element in {@link #dirtyPages} + the previous value in
this array. */
+ private final int[] sizes;
+
+ /**
+ * Private constructor.
+ */
+ private CheckpointDirtyPagesQueue() {
+ int size = 0;
+
+ int[] sizes = new int[dirtyPages.size()];
+
+ for (int i = 0; i < dirtyPages.size(); i++) {
+ sizes[i] = size += dirtyPages.get(i).getValue().size();
+ }
+
+ this.sizes = sizes;
+ }
+
+ /**
+ * Returns {@link true} if the next element of the queue was obtained.
+ *
+ * @param result Holder is the result of getting the next dirty page.
+ */
+ public boolean next(QueueResult result) {
+ int queuePosition = this.position.getAndIncrement();
+
+ if (queuePosition >= dirtyPagesCount) {
+ result.owner = null;
+
+ return false;
+ }
+
+ if (result.owner != this) {
+ result.owner = this;
+ result.index = 0;
+ }
+
+ int index = result.index;
+
+ if (queuePosition >= sizes[index]) {
+ if (queuePosition == sizes[index]) {
+ index++;
+ } else {
+ index = findDirtyPagesIndex(index, queuePosition);
+ }
+ }
+
+ result.index = index;
+ result.position = index > 0 ? queuePosition - sizes[index - 1] :
queuePosition;
+
+ return true;
+ }
+
+ /**
+ * Returns {@link true} if the queue is empty.
+ */
+ public boolean isEmpty() {
+ return position.get() >= dirtyPagesCount;
+ }
+
+ /**
+ * Returns the size of the queue.
+ */
+ public int size() {
+ return dirtyPagesCount - Math.min(dirtyPagesCount, position.get());
+ }
+
+ private int findDirtyPagesIndex(int index, int position) {
+ return Math.abs(Arrays.binarySearch(sizes, index, sizes.length,
position) + 1);
+ }
+
+ private CheckpointDirtyPages owner() {
+ return CheckpointDirtyPages.this;
+ }
+ }
+
+ /**
+ * View of {@link CheckpointDirtyPages} in which all dirty pages will
refer to the same {@link PersistentPageMemory} and contain the
+ * same groupId and partitionId and increasing pageIdx.
+ *
+ * <p>Thread safe.
+ */
+ class CheckpointDirtyPagesView {
+ /** Element index in {@link CheckpointDirtyPages#dirtyPages}. */
+ private final int index;
+
+ /** Starting position (inclusive) of the dirty page within the element
at {@link #index}. */
+ private final int fromPosition;
+
+ /** End position (inclusive) of the dirty page within the element at
{@link #index}. */
+ private final int toPosition;
+
+ /**
+ * Private constructor.
+ *
+ * @param index Element index in {@link
CheckpointDirtyPages#dirtyPages}.
+ * @param fromPosition Starting position (inclusive) of the dirty page
within the element at {@link #index}.
+ * @param toPosition End position (inclusive) of the dirty page within
the element at {@link #index}.
+ */
+ private CheckpointDirtyPagesView(int index, int fromPosition, int
toPosition) {
+ this.index = index;
+ this.fromPosition = fromPosition;
+ this.toPosition = toPosition;
+ }
+
+ /**
+ * Returns the dirty page by index.
+ *
+ * @param index Dirty page index.
+ */
+ public FullPageId get(int index) {
+ return dirtyPages.get(this.index).getValue().get(fromPosition +
index);
+ }
+
+ /**
+ * Returns the page memory for view.
+ */
+ public PersistentPageMemory pageMemory() {
+ return dirtyPages.get(index).getKey();
+ }
+
+ /**
+ * Returns the size of the view.
+ */
+ public int size() {
+ return toPosition - fromPosition + 1;
+ }
+
+ private CheckpointDirtyPages owner() {
+ return CheckpointDirtyPages.this;
+ }
+
+ private boolean isToPositionLast() {
+ return toPosition == dirtyPages.get(index).getValue().size() - 1;
+ }
+ }
+
+ /**
+ * Holder is the result of getting the next dirty page in {@link
CheckpointDirtyPagesQueue#next(QueueResult)}.
+ *
+ * <p>Not thread safe.
+ */
+ static class QueueResult {
+ private @Nullable CheckpointDirtyPagesQueue owner;
+
+ /** Element index in {@link CheckpointDirtyPages#dirtyPages}. */
+ private int index;
+
+ /** Position of the dirty page within the element at {@link #index}. */
+ private int position;
+
+ /**
+ * Returns the page memory for the associated dirty page.
+ */
+ public @Nullable PersistentPageMemory pageMemory() {
+ return owner == null ? null :
owner.owner().dirtyPages.get(index).getKey();
+ }
+
+ /**
+ * Returns dirty page.
+ */
+ public @Nullable FullPageId dirtyPage() {
+ return owner == null ? null :
owner.owner().dirtyPages.get(index).getValue().get(position);
+ }
+ }
+
+ private static boolean equalsByGroupAndPartition(FullPageId pageId0,
FullPageId pageId1) {
+ return pageId0.groupId() == pageId1.groupId() &&
partitionId(pageId0.pageId()) == partitionId(pageId1.pageId());
Review Comment:
How about introducing `FullPageId#partition()` method? It could be useful in
other places as well.
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemory.java:
##########
@@ -2095,6 +2095,7 @@ public Collection<FullPageId>
beginCheckpoint(CompletableFuture<?> allowToReplac
safeToUpdate.set(true);
+ // Less memory and no need for Set features.
Review Comment:
A couple of other things that are related to code NOT touched by this PR,
but still it might be useful.
1. `CollectionUtils#union()` returns a `Collection` that, it seems, might
calculate size incorrectly if it unites `Set`s (if you unite `Set.of(1)` with
another `Set.of(1)`, the result will report size 2, even though, for sets, it
should be 1). If this is how it is intended to work, I think it makes sense to
rename the method, because `union()` strongly suggests that it has 'set union'
semantics for sets. Also, this behavior should be clarified in the method
Javadoc.
2. `CollectionUtils#union()` does not make copies of the original
collections, it just makes a wrapper around them. This might be a surprise for
a user. I suggest to specifically mention this 'wrapping' mode in the method
Javadoc to avoid confusion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
