IGNITE-24239 Create a reproducer for IGNITE-23588

modules/page-memory/build.gradle

@@ -52,6 +52,7 @@ dependencies {
     testFixturesImplementation project(':ignite-core')
     testFixturesImplementation project(':ignite-failure-handler')
     testFixturesImplementation(testFixtures(project(':ignite-core')))
+    testFixturesImplementation libs.fastutil.core
     testFixturesImplementation libs.mockito.core
     testFixturesImplementation libs.auto.service.annotations
     testFixturesImplementation libs.jetbrains.annotations

modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/AbstractBplusTreePageMemoryTest.java

@@ -35,7 +35,7 @@
 import static org.apache.ignite.internal.testframework.IgniteTestUtils.runMultiThreaded;
 import static org.apache.ignite.internal.testframework.IgniteTestUtils.runMultiThreadedAsync;
 import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
-import static org.apache.ignite.internal.util.Constants.GiB;
+import static org.apache.ignite.internal.util.Constants.MiB;
 import static org.apache.ignite.internal.util.StringUtils.hexLong;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
@@ -79,6 +79,7 @@
 import java.util.function.Predicate;
 import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
 import org.apache.ignite.internal.lang.IgniteStringBuilder;
+import org.apache.ignite.internal.lang.IgniteSystemProperties;
 import org.apache.ignite.internal.logger.IgniteLogger;
 import org.apache.ignite.internal.pagememory.FullPageId;
 import org.apache.ignite.internal.pagememory.PageMemory;
@@ -109,6 +110,8 @@
  * An abstract class for testing {@link BplusTree} using different implementations of {@link PageMemory}.
  */
 public abstract class AbstractBplusTreePageMemoryTest extends BaseIgniteAbstractTest {
+    protected static final String BPLUS_TREE_TEST_SEED = "BPLUS_TREE_TEST_SEED";
+
     private static final short LONG_INNER_IO = 30000;
 
     private static final short LONG_LEAF_IO = 30001;
@@ -129,12 +132,12 @@ public abstract class AbstractBplusTreePageMemoryTest extends BaseIgniteAbstract
 
     protected static final int PAGE_SIZE = 512;
 
-    protected static final long MAX_MEMORY_SIZE = GiB;
+    protected static final long MAX_MEMORY_SIZE = 64 * MiB;
 
     /** Forces printing lock/unlock events on the test tree. */
     private static boolean PRINT_LOCKS = false;
 
-    private static final Collection<Long> rmvdIds = ConcurrentHashMap.newKeySet();
+    protected static final Collection<Long> rmvdIds = ConcurrentHashMap.newKeySet();
 
     @Nullable
     protected PageMemory pageMem;
@@ -155,7 +158,7 @@ public abstract class AbstractBplusTreePageMemoryTest extends BaseIgniteAbstract
     protected void beforeEach() throws Exception {
         stop.set(false);
 
-        long seed = System.nanoTime();
+        long seed = IgniteSystemProperties.getLong(BPLUS_TREE_TEST_SEED, System.nanoTime());
 
         println("Test seed: " + seed + "L; // ");
 
@@ -208,6 +211,9 @@ protected void afterTest() throws Exception {
             RMV_INC = -1;
             CNT = 10;
         }
+
+        TestPageLockListener.clearStaticResources();
+        rmvdIds.clear();
     }
 
     /**
@@ -2775,7 +2781,7 @@ private TestTree createTestTree(boolean canGetRow, AtomicLong globalRmvId) throw
         return tree;
     }
 
-    private TestTree createTestTree(boolean canGetRow) throws Exception {
+    protected TestTree createTestTree(boolean canGetRow) throws Exception {
         return createTestTree(canGetRow, new AtomicLong());
     }
 
@@ -3102,6 +3108,14 @@ private static class TestPageLockListener implements PageLockListener {
 
         private final PageLockListener delegate;
 
+        static void clearStaticResources() {
+            beforeReadLock.clear();
+            readLocks.clear();
+
+            beforeWriteLock.clear();
+            writeLocks.clear();
+        }
+
         /**
          * Constructor.
          *

modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/persistence/ItBplusTreePersistentPageMemoryTest.java

@@ -18,12 +18,17 @@
 package org.apache.ignite.internal.pagememory.tree.persistence;
 
 import static org.apache.ignite.internal.configuration.ConfigurationTestUtils.fixConfiguration;
+import static org.apache.ignite.internal.pagememory.datastructure.DataStructure.rnd;
 import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointTestUtils.mockCheckpointTimeoutLock;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.runMultiThreaded;
 import static org.apache.ignite.internal.util.Constants.MiB;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
+import java.util.concurrent.atomic.AtomicLongArray;
 import java.util.stream.LongStream;
 import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
 import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
+import org.apache.ignite.internal.lang.IgniteSystemProperties;
 import org.apache.ignite.internal.pagememory.PageMemory;
 import org.apache.ignite.internal.pagememory.TestPageIoRegistry;
 import org.apache.ignite.internal.pagememory.configuration.schema.PersistentPageMemoryProfileConfiguration;
@@ -33,15 +38,23 @@
 import org.apache.ignite.internal.pagememory.reuse.ReuseList;
 import org.apache.ignite.internal.pagememory.tree.AbstractBplusTreePageMemoryTest;
 import org.apache.ignite.internal.pagememory.tree.BplusTree;
+import org.apache.ignite.internal.pagememory.util.SequencedOffheapReadWriteLock;
 import org.apache.ignite.internal.storage.configurations.StorageProfileConfiguration;
+import org.apache.ignite.internal.testframework.WithSystemProperty;
+import org.apache.ignite.internal.util.OffheapReadWriteLock;
 import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
 /**
  * Class to test the {@link BplusTree} with {@link PersistentPageMemory}.
  */
 @ExtendWith(ConfigurationExtension.class)
 public class ItBplusTreePersistentPageMemoryTest extends AbstractBplusTreePageMemoryTest {
+    /** Dictates the implementation of {@link OffheapReadWriteLock} that will be used in a test. */
+    private static final String USE_SEQUENCED_RW_LOCK = "USE_SEQUENCED_RW_LOCK";
+
     @InjectConfiguration(
             polymorphicExtensions = { PersistentPageMemoryProfileConfigurationSchema.class },
             value = "mock = {"
@@ -51,13 +64,19 @@ public class ItBplusTreePersistentPageMemoryTest extends AbstractBplusTreePageMe
     )
     private StorageProfileConfiguration storageProfileCfg;
 
+    private OffheapReadWriteLock offheapReadWriteLock;
+
     /** {@inheritDoc} */
     @Override
     protected PageMemory createPageMemory() {
         TestPageIoRegistry ioRegistry = new TestPageIoRegistry();
 
         ioRegistry.loadFromServiceLoader();
 
+        offheapReadWriteLock = IgniteSystemProperties.getBoolean(USE_SEQUENCED_RW_LOCK)
+                ? new SequencedOffheapReadWriteLock()
+                : new OffheapReadWriteLock(128);
+
         return new PersistentPageMemory(
                 (PersistentPageMemoryProfileConfiguration) fixConfiguration(storageProfileCfg),
                 ioRegistry,
@@ -69,8 +88,115 @@ protected PageMemory createPageMemory() {
                 (fullPageId, buf, tag) -> {
                 },
                 mockCheckpointTimeoutLock(true),
-                PAGE_SIZE
+                PAGE_SIZE,
+                offheapReadWriteLock
+        );
+    }
+
+    /**
+     * Test is based on {@link AbstractBplusTreePageMemoryTest#testMassiveRemove2_true()}, but uses a deterministic execution order in order
+     * to achieve a 100% reliable fail rate.
+     *
+     * @throws Exception If failed.
+     */
+    @Disabled("https://issues.apache.org/jira/browse/IGNITE-23588")
+    @Test
+    @WithSystemProperty(key = USE_SEQUENCED_RW_LOCK, value = "true")
+    @WithSystemProperty(key = BPLUS_TREE_TEST_SEED, value = "1161542256747481")
+    public void testMassiveRemoveCorruption() throws Exception {
+        SequencedOffheapReadWriteLock offheapReadWriteLock = (SequencedOffheapReadWriteLock) this.offheapReadWriteLock;
+
+        //noinspection AssignmentToStaticFieldFromInstanceMethod
+        MAX_PER_PAGE = 2;
+        int threads = 16;
+        int batch = 500;
+        boolean canGetRow = true;
+
+        println("[maxPerPage=" + MAX_PER_PAGE
+                + ", threads=" + threads
+                + ", batch=" + batch
+                + ", canGetRow=" + canGetRow
+                + "]"
         );
+
+        int keys = threads * batch;
+
+        TestTree tree = createTestTree(canGetRow);
+
+        // Inverted insertion order, like in original test.
+        for (long k = keys - 1; k >= 0; k--) {
+            tree.put(k);
+        }
+
+        assertEquals(keys, tree.size());
+        tree.validateTree();
+
+        AtomicLongArray rmvd = new AtomicLongArray(keys);
+
+        // All the random bits are pre-calculated in advance.
+        // It doesn't have to be this way, because there's a critical section in the loop later anyway. But, this is the version of the code
+        // with a known seed, so let's leave it as it is right now.
+        int[][] rndEx = new int[threads][keys];
+        for (int i = 0; i < rndEx.length; i++) {
+            for (int j = 0; j < rndEx[i].length; j++) {
+                rndEx[i][j] = rnd.nextInt(keys);
+            }
+        }
+
+        offheapReadWriteLock.startSequencing(() -> rnd.nextInt(threads), threads);
+
+        String threadNamePrefix = "remove";
+        runMultiThreaded(() -> {
+            // Here we rely on a thread naming convention in "runMultiThreaded" to get thread's ID.
+            int threadId = Integer.parseInt(Thread.currentThread().getName().substring(threadNamePrefix.length()));
+
+            offheapReadWriteLock.setCurrentThreadId(threadId);
+
+            try {
+                int rndIdx = 0;
+                while (true) {
+                    int idx = 0;
+                    boolean found = false;
+
+                    offheapReadWriteLock.await();
+                    try {
+                        int shift = rndEx[threadId][rndIdx++];
+                        // Since this loop modifies a shared resource ("rmvd"), we must have a critical section here as well.
+                        for (int i = 0; i < keys; i++) {
+                            idx = (i + shift) % keys;
+
+                            if (rmvd.get(idx) == 0 && rmvd.compareAndSet(idx, 0, 1)) {
+                                found = true;
+
+                                break;
+                            }
+                        }
+                    } finally {
+                        offheapReadWriteLock.release();
+                    }
+
+                    if (!found) {
+                        break;
+                    }
+
+                    Long key = (long) idx;
+                    assertEquals(key, tree.remove(key));
+
+                    if (canGetRow) {
+                        rmvdIds.add(key);
+                    }
+                }
+            } finally {
+                offheapReadWriteLock.complete();
+            }
+
+            return null;
+        }, threads, threadNamePrefix);
+
+        offheapReadWriteLock.stopSequencing();
+
+        assertEquals(0, tree.size());
+        tree.validateTree();
     }
 
     /** {@inheritDoc} */

modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/persistence/ItBplusTreeReuseListPersistentPageMemoryTest.java

@@ -34,6 +34,7 @@
 import org.apache.ignite.internal.pagememory.persistence.TestPageReadWriteManager;
 import org.apache.ignite.internal.pagememory.tree.AbstractBplusTreeReusePageMemoryTest;
 import org.apache.ignite.internal.storage.configurations.StorageProfileConfiguration;
+import org.apache.ignite.internal.util.OffheapReadWriteLock;
 import org.junit.jupiter.api.extension.ExtendWith;
 
 /**
@@ -70,7 +71,8 @@ protected PageMemory createPageMemory() throws Exception {
                 (fullPageId, buf, tag) -> {
                 },
                 mockCheckpointTimeoutLock(true),
-                PAGE_SIZE
+                PAGE_SIZE,
+                new OffheapReadWriteLock(128)
         );
     }