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 2023/10/16 18:47:35 UTC
[ignite-3] branch main updated: IGNITE-20657 At the checkpoint ArrayIndexOutOfBoundsException (#2696)
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 c962aa1bbb IGNITE-20657 At the checkpoint ArrayIndexOutOfBoundsException (#2696)
c962aa1bbb is described below
commit c962aa1bbb94e73a4d8ce2403aad3d629dd55666
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Mon Oct 16 21:47:29 2023 +0300
IGNITE-20657 At the checkpoint ArrayIndexOutOfBoundsException (#2696)
---
.../persistence/store/FilePageStore.java | 121 ++++++++++++---------
.../persistence/store/FilePageStoreTest.java | 16 +++
2 files changed, 86 insertions(+), 51 deletions(-)
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStore.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStore.java
index 7d1c9cb991..041c7c861b 100644
--- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStore.java
+++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStore.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.pagememory.persistence.store;
+import static java.util.Collections.unmodifiableList;
import static org.apache.ignite.internal.pagememory.util.PageIdUtils.pageIndex;
import java.io.IOException;
@@ -24,12 +25,11 @@ import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.nio.ByteBuffer;
import java.nio.file.Path;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
-import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.IntFunction;
import java.util.function.Supplier;
import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
@@ -53,6 +53,8 @@ public class FilePageStore implements PageStore {
private static final VarHandle NEW_DELTA_FILE_PAGE_STORE_IO_FUTURE;
+ private static final VarHandle DELTA_FILE_PAGE_STORE_IOS;
+
static {
try {
PAGE_COUNT = MethodHandles.lookup().findVarHandle(FilePageStore.class, "pageCount", int.class);
@@ -62,6 +64,8 @@ public class FilePageStore implements PageStore {
"newDeltaFilePageStoreIoFuture",
CompletableFuture.class
);
+
+ DELTA_FILE_PAGE_STORE_IOS = MethodHandles.lookup().findVarHandle(FilePageStore.class, "deltaFilePageStoreIos", List.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
@@ -88,8 +92,8 @@ public class FilePageStore implements PageStore {
/** New page allocation listener. */
private volatile @Nullable PageAllocationListener pageAllocationListener;
- /** Delta file page store IOs. */
- private final List<DeltaFilePageStoreIo> deltaFilePageStoreIos;
+ /** Delta file page store IOs. Copy-on-write list. */
+ private volatile List<DeltaFilePageStoreIo> deltaFilePageStoreIos;
/** Future with a new delta file page store. */
private volatile @Nullable CompletableFuture<DeltaFilePageStoreIo> newDeltaFilePageStoreIoFuture;
@@ -112,10 +116,9 @@ public class FilePageStore implements PageStore {
}
this.filePageStoreIo = filePageStoreIo;
- this.deltaFilePageStoreIos = new CopyOnWriteArrayList<>(Arrays.asList(deltaFilePageStoreIos));
+ this.deltaFilePageStoreIos = Arrays.asList(deltaFilePageStoreIos);
}
- /** {@inheritDoc} */
@Override
public void stop(boolean clean) throws IgniteInternalCheckedException {
filePageStoreIo.stop(clean);
@@ -125,7 +128,6 @@ public class FilePageStore implements PageStore {
}
}
- /** {@inheritDoc} */
@Override
public int allocatePage() throws IgniteInternalCheckedException {
ensure();
@@ -141,7 +143,6 @@ public class FilePageStore implements PageStore {
return pageIdx;
}
- /** {@inheritDoc} */
@Override
public int pages() {
return pageCount;
@@ -181,7 +182,6 @@ public class FilePageStore implements PageStore {
filePageStoreIo.read(pageId, filePageStoreIo.pageOffset(pageId), pageBuf, keepCrc);
}
- /** {@inheritDoc} */
@Override
public void read(long pageId, ByteBuffer pageBuf, boolean keepCrc) throws IgniteInternalCheckedException {
assert pageIndex(pageId) <= pageCount : "pageIdx=" + pageIndex(pageId) + ", pageCount=" + pageCount;
@@ -189,7 +189,6 @@ public class FilePageStore implements PageStore {
readWithoutPageIdCheck(pageId, pageBuf, keepCrc);
}
- /** {@inheritDoc} */
@Override
public void write(long pageId, ByteBuffer pageBuf, boolean calculateCrc) throws IgniteInternalCheckedException {
assert pageIndex(pageId) <= pageCount : "pageIdx=" + pageIndex(pageId) + ", pageCount=" + pageCount;
@@ -197,25 +196,21 @@ public class FilePageStore implements PageStore {
filePageStoreIo.write(pageId, pageBuf, calculateCrc);
}
- /** {@inheritDoc} */
@Override
public void sync() throws IgniteInternalCheckedException {
filePageStoreIo.sync();
}
- /** {@inheritDoc} */
@Override
public boolean exists() {
return filePageStoreIo.exists();
}
- /** {@inheritDoc} */
@Override
public void ensure() throws IgniteInternalCheckedException {
filePageStoreIo.ensure();
}
- /** {@inheritDoc} */
@Override
public void close() throws IOException {
filePageStoreIo.close();
@@ -266,12 +261,12 @@ public class FilePageStore implements PageStore {
*
* <p>Thread safe.
*
- * @param deltaFilPathFunction Function to get the path to the delta file page store, the argument is the index of the delta file.
+ * @param deltaFilePathFunction Function to get the path to the delta file page store, the argument is the index of the delta file.
* @param pageIndexesSupplier Page indexes supplier that will only be called if the current thread creates a delta file page store.
* @return Future that will be completed when the new delta file page store is created.
*/
public CompletableFuture<DeltaFilePageStoreIo> getOrCreateNewDeltaFile(
- IntFunction<Path> deltaFilPathFunction,
+ IntFunction<Path> deltaFilePathFunction,
Supplier<int[]> pageIndexesSupplier
) {
CompletableFuture<DeltaFilePageStoreIo> future = this.newDeltaFilePageStoreIoFuture;
@@ -285,27 +280,47 @@ public class FilePageStore implements PageStore {
return newDeltaFilePageStoreIoFuture;
}
- int nextIndex = deltaFilePageStoreIos.isEmpty() ? 0 : deltaFilePageStoreIos.get(0).fileIndex() + 1;
+ DeltaFilePageStoreIo newDeltaFilePageStoreIo = addNewDeltaFilePageStoreIoToHead(pageIndexesSupplier.get(), deltaFilePathFunction);
- DeltaFilePageStoreIoHeader header = new DeltaFilePageStoreIoHeader(
- LATEST_DELTA_FILE_PAGE_STORE_VERSION,
- nextIndex,
- filePageStoreIo.pageSize(),
- pageIndexesSupplier.get()
- );
+ future.complete(newDeltaFilePageStoreIo);
- DeltaFilePageStoreIo deltaFilePageStoreIo = new DeltaFilePageStoreIo(
- filePageStoreIo.ioFactory,
- deltaFilPathFunction.apply(nextIndex),
- header
- );
+ return future;
+ }
- // Should add to the head, since read operations should always start from the most recent.
- deltaFilePageStoreIos.add(0, deltaFilePageStoreIo);
+ private DeltaFilePageStoreIo addNewDeltaFilePageStoreIoToHead(int[] pageIndexes, IntFunction<Path> deltaFilePathFunction) {
+ List<DeltaFilePageStoreIo> previousValue;
+ List<DeltaFilePageStoreIo> newValue;
- future.complete(deltaFilePageStoreIo);
+ DeltaFilePageStoreIo newDeltaFilePageStoreIo;
- return future;
+ do {
+ previousValue = deltaFilePageStoreIos;
+
+ int nextIndex = previousValue.isEmpty() ? 0 : previousValue.get(0).fileIndex() + 1;
+
+ var header = new DeltaFilePageStoreIoHeader(
+ LATEST_DELTA_FILE_PAGE_STORE_VERSION,
+ nextIndex,
+ filePageStoreIo.pageSize(),
+ pageIndexes
+ );
+
+ newDeltaFilePageStoreIo = new DeltaFilePageStoreIo(
+ filePageStoreIo.ioFactory,
+ deltaFilePathFunction.apply(nextIndex),
+ header
+ );
+
+ newValue = new ArrayList<>(previousValue.size() + 1);
+
+ // Should add to the head, since read operations should always start from the most recent.
+ newValue.add(newDeltaFilePageStoreIo);
+ newValue.addAll(previousValue);
+
+ newValue = unmodifiableList(newValue);
+ } while (!DELTA_FILE_PAGE_STORE_IOS.compareAndSet(this, previousValue, newValue));
+
+ return newDeltaFilePageStoreIo;
}
/**
@@ -340,43 +355,47 @@ public class FilePageStore implements PageStore {
}
/**
- * Returns the delta file to compaction (oldest).
+ * Returns the delta file to compaction (oldest), {@code null} if there is nothing to compact.
*
* <p>Thread safe.
*/
public @Nullable DeltaFilePageStoreIo getDeltaFileToCompaction() {
// Snapshot of delta files.
- Iterator<DeltaFilePageStoreIo> iterator = deltaFilePageStoreIos.iterator();
-
- // Last one is the oldest.
- DeltaFilePageStoreIo last = null;
-
- int count = 0;
-
- while (iterator.hasNext()) {
- last = iterator.next();
+ List<DeltaFilePageStoreIo> deltaFilePageStoreIos = this.deltaFilePageStoreIos;
- count++;
+ // If there are no delta files or if last is just created, then it cannot be compacted yet.
+ if (deltaFilePageStoreIos.isEmpty() || (deltaFilePageStoreIos.size() == 1 && newDeltaFilePageStoreIoFuture != null)) {
+ return null;
}
- // If last is just created, then it cannot be compacted yet.
- if (count == 1 && newDeltaFilePageStoreIoFuture != null) {
- last = null;
- }
-
- return last;
+ return deltaFilePageStoreIos.get(deltaFilePageStoreIos.size() - 1);
}
/**
* Deletes delta file.
*
- * <p>Thread safe.
+ * <p>Thread safe.</p>
*
* @param deltaFilePageStoreIo Delta file to be deleted.
* @return {@code True} if the delta file being removed was present.
*/
public boolean removeDeltaFile(DeltaFilePageStoreIo deltaFilePageStoreIo) {
- return deltaFilePageStoreIos.remove(deltaFilePageStoreIo);
+ List<DeltaFilePageStoreIo> previousValue;
+ List<DeltaFilePageStoreIo> newValue;
+
+ boolean removed;
+
+ do {
+ previousValue = deltaFilePageStoreIos;
+
+ newValue = new ArrayList<>(previousValue);
+
+ removed = newValue.remove(deltaFilePageStoreIo);
+
+ newValue = unmodifiableList(newValue);
+ } while (!DELTA_FILE_PAGE_STORE_IOS.compareAndSet(this, previousValue, newValue));
+
+ return removed;
}
/**
diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreTest.java
index 0b5028b4a6..f7edec9386 100644
--- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreTest.java
+++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreTest.java
@@ -24,6 +24,8 @@ import static org.apache.ignite.internal.pagememory.persistence.store.FilePageSt
import static org.apache.ignite.internal.pagememory.persistence.store.TestPageStoreUtils.arr;
import static org.apache.ignite.internal.pagememory.persistence.store.TestPageStoreUtils.createDataPageId;
import static org.apache.ignite.internal.pagememory.persistence.store.TestPageStoreUtils.createPageByteBuffer;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.runRace;
+import static org.apache.ignite.internal.util.ArrayUtils.INT_EMPTY_ARRAY;
import static org.apache.ignite.internal.util.GridUnsafe.allocateBuffer;
import static org.apache.ignite.internal.util.GridUnsafe.freeBuffer;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -54,6 +56,7 @@ import org.apache.ignite.internal.pagememory.io.PageIo;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -504,6 +507,19 @@ public class FilePageStoreTest extends BaseIgniteAbstractTest {
}
}
+ @RepeatedTest(100)
+ void testConcurrentAddNewAndCompactDeltaFile() throws Exception {
+ try (FilePageStore filePageStore = createFilePageStore(workDir.resolve("test"), mock(DeltaFilePageStoreIo.class))) {
+ DeltaFilePageStoreIo deltaFileToCompaction = filePageStore.getDeltaFileToCompaction();
+ Path newDeltaFilePath = workDir.resolve("test_2");
+
+ runRace(
+ () -> filePageStore.getOrCreateNewDeltaFile(newDeltaFileIndex -> newDeltaFilePath, () -> INT_EMPTY_ARRAY),
+ () -> filePageStore.removeDeltaFile(deltaFileToCompaction)
+ );
+ }
+ }
+
private static FilePageStore createFilePageStore(Path filePath) {
return createFilePageStore(filePath, new FilePageStoreHeader(VERSION_1, PAGE_SIZE));
}