diff --git a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/AbstractBplusTreePageMemoryTest.java b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/AbstractBplusTreePageMemoryTest.java index b014c5154aff..046d6d461c52 100644 --- a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/AbstractBplusTreePageMemoryTest.java +++ b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/AbstractBplusTreePageMemoryTest.java @@ -82,6 +82,7 @@ import org.apache.ignite.internal.logger.IgniteLogger; import org.apache.ignite.internal.pagememory.FullPageId; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.TestPageIoRegistry; import org.apache.ignite.internal.pagememory.datastructure.DataStructure; import org.apache.ignite.internal.pagememory.io.IoVersions; @@ -142,6 +143,9 @@ public abstract class AbstractBplusTreePageMemoryTest extends BaseIgniteAbstract @Nullable protected PageMemory pageMem; + @Nullable + protected PartitionPageMemory partitionPageMemory; + @Nullable private ReuseList reuseList; @@ -165,8 +169,9 @@ protected void beforeEach() throws Exception { rnd = new Random(seed); pageMem = createPageMemory(); + partitionPageMemory = pageMem.createPartitionPageMemory(GROUP_ID, 0); - reuseList = createReuseList(GROUP_ID, 0, pageMem, 0, true); + reuseList = createReuseList(GROUP_ID, 0, partitionPageMemory, 0, true); } @AfterEach @@ -234,7 +239,7 @@ protected long getTestTimeout() { protected abstract @Nullable ReuseList createReuseList( int grpId, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, long rootId, boolean initNew ) throws Exception; @@ -2568,7 +2573,7 @@ private static void findNextForKeys(TestTree tree, long keyCount) throws Excepti } private TestTree reCreateTestTree(TestTree tree, long globalRmvId) throws Exception { - return new TestTree(tree.metaFullPageId(), reuseList, tree.canGetRow, pageMem, new AtomicLong(globalRmvId), false); + return new TestTree(tree.metaFullPageId(), reuseList, tree.canGetRow, partitionPageMemory, new AtomicLong(globalRmvId), false); } private void doTestRandomPutRemoveMultithreaded(boolean canGetRow) throws Exception { @@ -2762,7 +2767,7 @@ private static int size(Cursor c) { } private TestTree createTestTree(boolean canGetRow, AtomicLong globalRmvId) throws Exception { - var tree = new TestTree(allocateMetaPage(), reuseList, canGetRow, pageMem, globalRmvId, true); + var tree = new TestTree(allocateMetaPage(), reuseList, canGetRow, partitionPageMemory, globalRmvId, true); assertEquals(0, tree.size()); assertEquals(0, tree.rootLevel()); @@ -2775,7 +2780,7 @@ protected TestTree createTestTree(boolean canGetRow) throws Exception { } private FullPageId allocateMetaPage() throws Exception { - return new FullPageId(pageMem.allocatePage(reuseList, GROUP_ID, 0, FLAG_AUX), GROUP_ID); + return new FullPageId(partitionPageMemory.allocatePage(reuseList, GROUP_ID, 0, FLAG_AUX), GROUP_ID); } /** Test tree. */ @@ -2802,7 +2807,7 @@ protected static class TestTree extends BplusTree { FullPageId metaPageId, @Nullable ReuseList reuseList, boolean canGetRow, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, boolean initNew ) throws Exception { diff --git a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/AbstractBplusTreeReusePageMemoryTest.java b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/AbstractBplusTreeReusePageMemoryTest.java index 5cb286168ccf..6a86cbc3b7ee 100644 --- a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/AbstractBplusTreeReusePageMemoryTest.java +++ b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/AbstractBplusTreeReusePageMemoryTest.java @@ -19,6 +19,7 @@ import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.freelist.FreeListImpl; import org.apache.ignite.internal.pagememory.reuse.ReuseList; @@ -31,7 +32,7 @@ public abstract class AbstractBplusTreeReusePageMemoryTest extends AbstractBplus protected ReuseList createReuseList( int grpId, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, long rootId, boolean initNew ) throws IgniteInternalCheckedException { diff --git a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/ItBplusTreeReplaceRemoveRaceTest.java b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/ItBplusTreeReplaceRemoveRaceTest.java index fef09030615e..7785db2ff47e 100644 --- a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/ItBplusTreeReplaceRemoveRaceTest.java +++ b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/ItBplusTreeReplaceRemoveRaceTest.java @@ -34,6 +34,7 @@ import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.pagememory.FullPageId; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.TestPageIoRegistry; import org.apache.ignite.internal.pagememory.configuration.VolatileDataRegionConfiguration; import org.apache.ignite.internal.pagememory.inmemory.VolatilePageMemory; @@ -67,9 +68,13 @@ public class ItBplusTreeReplaceRemoveRaceTest extends BaseIgniteAbstractTest { @Nullable protected PageMemory pageMem; + @Nullable + protected PartitionPageMemory partitionPageMemory; + @BeforeEach protected void beforeEach() throws Exception { pageMem = createPageMemory(); + partitionPageMemory = pageMem.createPartitionPageMemory(GROUP_ID, 0); } @AfterEach @@ -92,7 +97,7 @@ protected PageMemory createPageMemory() throws Exception { } private FullPageId allocateMetaPage() throws IgniteInternalCheckedException { - return new FullPageId(pageMem.allocatePageNoReuse(GROUP_ID, 0, FLAG_AUX), GROUP_ID); + return new FullPageId(partitionPageMemory.allocatePageNoReuse(GROUP_ID, 0, FLAG_AUX), GROUP_ID); } /** @@ -126,7 +131,7 @@ protected static class TestPairTree extends BplusTree { */ public TestPairTree( FullPageId metaPageId, - PageMemory pageMem + PartitionPageMemory pageMem ) throws IgniteInternalCheckedException { super( "test", @@ -419,7 +424,7 @@ public void testConcurrentPutRemoveSameRow() throws Exception { * @throws Exception If failed. */ private TestPairTree prepareBplusTree() throws Exception { - TestPairTree tree = new TestPairTree(allocateMetaPage(), pageMem); + TestPairTree tree = new TestPairTree(allocateMetaPage(), partitionPageMemory); tree.putx(new Pair(1, 0)); tree.putx(new Pair(2, 0)); diff --git a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/inmemory/ItBplusTreeFakeReuseVolatilePageMemoryTest.java b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/inmemory/ItBplusTreeFakeReuseVolatilePageMemoryTest.java index 1654f4afd17d..1997fb6eec18 100644 --- a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/inmemory/ItBplusTreeFakeReuseVolatilePageMemoryTest.java +++ b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/inmemory/ItBplusTreeFakeReuseVolatilePageMemoryTest.java @@ -18,7 +18,7 @@ package org.apache.ignite.internal.pagememory.tree.inmemory; import java.util.concurrent.ConcurrentLinkedDeque; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.io.PageIo; import org.apache.ignite.internal.pagememory.reuse.ReuseBag; import org.apache.ignite.internal.pagememory.reuse.ReuseList; @@ -32,7 +32,7 @@ public class ItBplusTreeFakeReuseVolatilePageMemoryTest extends ItBplusTreeVolat protected ReuseList createReuseList( int grpId, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, long rootId, boolean initNew ) { diff --git a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/inmemory/ItBplusTreeVolatilePageMemoryTest.java b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/inmemory/ItBplusTreeVolatilePageMemoryTest.java index 4348bd20636c..7d60f8359e41 100644 --- a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/inmemory/ItBplusTreeVolatilePageMemoryTest.java +++ b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/inmemory/ItBplusTreeVolatilePageMemoryTest.java @@ -19,6 +19,7 @@ import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.TestPageIoRegistry; import org.apache.ignite.internal.pagememory.configuration.VolatileDataRegionConfiguration; import org.apache.ignite.internal.pagememory.inmemory.VolatilePageMemory; @@ -65,7 +66,7 @@ protected long acquiredPages() { protected @Nullable ReuseList createReuseList( int grpId, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, long rootId, boolean initNew ) { diff --git a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/persistence/ItBplusTreePersistentPageMemoryTest.java b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/persistence/ItBplusTreePersistentPageMemoryTest.java index 933d600f542a..e0d6da9e3db5 100644 --- a/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/persistence/ItBplusTreePersistentPageMemoryTest.java +++ b/modules/page-memory/src/integrationTest/java/org/apache/ignite/internal/pagememory/tree/persistence/ItBplusTreePersistentPageMemoryTest.java @@ -24,6 +24,7 @@ import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension; import org.apache.ignite.internal.lang.IgniteSystemProperties; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.TestPageIoRegistry; import org.apache.ignite.internal.pagememory.configuration.PersistentDataRegionConfiguration; import org.apache.ignite.internal.pagememory.persistence.PageHeader; @@ -92,7 +93,7 @@ protected long acquiredPages() { protected @Nullable ReuseList createReuseList( int grpId, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, long rootId, boolean initNew ) { diff --git a/modules/page-memory/src/jmh/java/org/apache/ignite/internal/pagememory/benchmark/PageReplacementBenchmark.java b/modules/page-memory/src/jmh/java/org/apache/ignite/internal/pagememory/benchmark/PageReplacementBenchmark.java index ac6b2cad70bf..17c580610469 100644 --- a/modules/page-memory/src/jmh/java/org/apache/ignite/internal/pagememory/benchmark/PageReplacementBenchmark.java +++ b/modules/page-memory/src/jmh/java/org/apache/ignite/internal/pagememory/benchmark/PageReplacementBenchmark.java @@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.TestPageIoModule.TestSimpleValuePageIo; import org.apache.ignite.internal.pagememory.configuration.ReplacementMode; import org.apache.ignite.internal.pagememory.io.PageIo; @@ -92,6 +93,7 @@ public class PageReplacementBenchmark extends PersistentPageMemoryBenchmarkBase public CachePressure cachePressure; private long[] pageIds; + private PartitionPageMemory[] partitionPageMemories; private int workingSetSize; @@ -262,6 +264,7 @@ public void setup(Blackhole blackhole) throws Exception { */ private void prepareWorkingSet() throws Exception { pageIds = new long[workingSetSize]; + partitionPageMemories = new PartitionPageMemory[workingSetSize]; // Allocate 80% of capacity at a time to avoid hitting dirty pages limit. int batchSize = (int) Math.round(REGION_CAPACITY_PAGES * 0.8); @@ -276,12 +279,13 @@ private void prepareWorkingSet() throws Exception { // Distribute pages randomly across partitions. for (int i = batchStart; i < batchEnd; i++) { int partitionId = partitionRandom.nextInt(PARTITION_COUNT); - pageIds[i] = persistentPageMemory().allocatePage(null, GROUP_ID, partitionId, FLAG_DATA); + PartitionPageMemory partitionPageMemory = partitionPageMemory(partitionId); + pageIds[i] = partitionPageMemory.allocatePage(null, GROUP_ID, partitionId, FLAG_DATA); + partitionPageMemories[i] = partitionPageMemory; } for (int i = batchStart; i < batchEnd; i++) { - long pageId = pageIds[i]; - writePage(pageId, pageIo); + writePage(i, pageIo); } } finally { checkpointManager().checkpointTimeoutLock().checkpointReadUnlock(); @@ -325,9 +329,8 @@ private void warmupCache(Blackhole blackhole) throws IgniteInternalCheckedExcept try { for (int i = 0; i < warmupPages; i++) { int index = warmupDistribution.next(); - long pageId = pageIds[index]; - accessPageReadOnly(pageId, 0, blackhole); + accessPageReadOnly(index, 0, blackhole); } } finally { checkpointManager().checkpointTimeoutLock().checkpointReadUnlock(); @@ -387,13 +390,15 @@ public void zipfianLatencyFourThreads(ThreadState state, Blackhole blackhole) private void benchmarkIteration(ThreadState state, Blackhole blackhole) throws IgniteInternalCheckedException { int index = state.nextZipfianIndex(); - long pageId = pageIds[index]; - accessPageReadOnly(pageId, state.threadIndex(), blackhole); + accessPageReadOnly(index, state.threadIndex(), blackhole); } - private void accessPageReadOnly(long pageId, int threadIndex, Blackhole blackhole) + private void accessPageReadOnly(int index, int threadIndex, Blackhole blackhole) throws IgniteInternalCheckedException { - long page = persistentPageMemory().acquirePage(GROUP_ID, pageId); + long pageId = pageIds[index]; + PartitionPageMemory partitionPageMemory = partitionPageMemories[index]; + + long page = partitionPageMemory.acquirePage(GROUP_ID, pageId); if (page == 0) { throw new IllegalStateException(String.format( @@ -408,22 +413,25 @@ private void accessPageReadOnly(long pageId, int threadIndex, Blackhole blackhol try { blackhole.consume(page); } finally { - persistentPageMemory().releasePage(GROUP_ID, pageId, page); + partitionPageMemory.releasePage(GROUP_ID, pageId, page); } } - private void writePage(long pageId, PageIo pageIo) throws IgniteInternalCheckedException { - long page = persistentPageMemory().acquirePage(GROUP_ID, pageId); + private void writePage(int index, PageIo pageIo) throws IgniteInternalCheckedException { + long pageId = pageIds[index]; + PartitionPageMemory partitionPageMemory = partitionPageMemories[index]; + + long page = partitionPageMemory.acquirePage(GROUP_ID, pageId); try { - long pageAddr = persistentPageMemory().writeLock(GROUP_ID, pageId, page); + long pageAddr = partitionPageMemory.writeLock(GROUP_ID, pageId, page); try { pageIo.initNewPage(pageAddr, pageId, PAGE_SIZE); TestSimpleValuePageIo.setLongValue(pageAddr, pageId); } finally { - persistentPageMemory().writeUnlock(GROUP_ID, pageId, page, true); + partitionPageMemory.writeUnlock(GROUP_ID, pageId, page, true); } } finally { - persistentPageMemory().releasePage(GROUP_ID, pageId, page); + partitionPageMemory.releasePage(GROUP_ID, pageId, page); } } diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageIdAllocator.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageIdAllocator.java index 39b16ae90aa1..b7a70e1c07d2 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageIdAllocator.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageIdAllocator.java @@ -31,6 +31,7 @@ /** * Class responsible for allocating/freeing page IDs. */ +// TODO IGNITE-28429 Remove "groupId" parameter from all methods. public interface PageIdAllocator { /** * Flag for a Data page. Also used by partition meta and tracking pages. This type doesn't use the Page ID rotation mechanism. diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageMemory.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageMemory.java index d65c490ebb32..9d29f3ee07c8 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageMemory.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageMemory.java @@ -17,20 +17,18 @@ package org.apache.ignite.internal.pagememory; -import java.nio.ByteBuffer; import org.apache.ignite.internal.lang.IgniteInternalException; -import org.apache.ignite.internal.pagememory.io.PageIo; -import org.apache.ignite.internal.pagememory.io.PageIoRegistry; /** * Class responsible for pages storage and handling. */ // TODO IGNITE-16350 Improve javadoc in this class. -public interface PageMemory extends PageIdAllocator, PageSupport { +// TODO IGNITE-28429 Remove the inheritance. +public interface PageMemory { /** * Stops page memory. * - * @param deallocate {@code True} to deallocate memory, {@code false} to allow memory reuse on subsequent {@link #start()} + * @param deallocate {@code True} to deallocate memory, {@code false} to allow memory reuse on subsequent {@code start()} */ void stop(boolean deallocate) throws IgniteInternalException; @@ -39,35 +37,22 @@ public interface PageMemory extends PageIdAllocator, PageSupport { */ int pageSize(); - /** - * Returns a page size without the encryption overhead, in bytes. - * - * @param groupId Group id. - */ - // TODO IGNITE-16350 Consider renaming. - int realPageSize(int groupId); - /** * Returns a page's size with system overhead, in bytes. */ // TODO IGNITE-16350 Consider renaming. int systemPageSize(); - /** - * Wraps a page address, obtained by a {@code readLock}/{@code writeLock} or their variants, into a direct byte buffer. - * - * @param pageAddr Page address. - * @return Page byte buffer. - */ - ByteBuffer pageBuffer(long pageAddr); - /** * Returns the total number of pages loaded into memory. */ long loadedPages(); /** - * Returns a registry to obtain {@link PageIo} instances for pages. + * Creates a new instance of {@link PartitionPageMemory} for a specified partition. + * + * @param groupId Group ID for the specific partition. + * @param partitionId Partition ID of the specific partition. */ - PageIoRegistry ioRegistry(); + PartitionPageMemory createPartitionPageMemory(int groupId, int partitionId); } diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageSupport.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageSupport.java index ad40cea47779..226c8f09de0b 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageSupport.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PageSupport.java @@ -23,6 +23,7 @@ * Class responsible for acquiring/releasing and locking/unlocking pages. */ // TODO IGNITE-16350 Document a naming convention for "page" and "pageAddr" parameters. +// TODO IGNITE-28429 Remove "groupId" parameter from all methods. public interface PageSupport { /** * Returns an absolute pointer to a page, associated with the given page ID. Each pointer obtained with this method must be released by @@ -83,6 +84,16 @@ public interface PageSupport { */ long writeLock(int groupId, long pageId, long page); + /** + * Acquired a write lock on the page, without checking the page tag. + * + * @param groupId Group ID. + * @param pageId Page ID. + * @param page Page pointer. + * @return Address of a buffer with contents of the given page or {@code 0L} if attempt to take the write lock failed. + */ + long writeLockForce(int groupId, long pageId, long page); + /** * Tries to acquire a write lock on the page. * @@ -102,14 +113,4 @@ public interface PageSupport { * @param dirtyFlag Determines whether the page was modified since the last checkpoint. */ void writeUnlock(int groupId, long pageId, long page, boolean dirtyFlag); - - /** - * Checks whether the page is dirty. - * - * @param groupId Group ID. - * @param pageId Page ID. - * @param page Page pointer. - * @return {@code True} if the page is dirty. - */ - boolean isDirty(int groupId, long pageId, long page); } diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PartitionPageMemory.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PartitionPageMemory.java new file mode 100644 index 000000000000..ae21079f4a55 --- /dev/null +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/PartitionPageMemory.java @@ -0,0 +1,55 @@ +/* + * 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; + +import org.apache.ignite.internal.pagememory.io.PageIo; +import org.apache.ignite.internal.pagememory.io.PageIoRegistry; + +/** + * A page memory instance that's supposed to be bound to a single partition only. Currently it's not the case, the job is split into phases. + */ +// TODO IGNITE-28429 Remove "groupId" parameter from all methods. +public interface PartitionPageMemory extends PageSupport, PageIdAllocator { + /** + * Returns a group ID associated with this instance. + */ + int groupId(); + + /** + * Returns a partition ID associated with this instance. + */ + int partitionId(); + + /** + * Returns a registry to obtain {@link PageIo} instances for pages. + */ + PageIoRegistry ioRegistry(); + + /** + * Returns a page's size in bytes. + */ + int pageSize(); + + /** + * Returns a page size without the encryption overhead, in bytes. + * + * @param groupId Group id. + */ + // TODO IGNITE-16350 Consider renaming. + int realPageSize(int groupId); +} diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datapage/DataPageReader.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datapage/DataPageReader.java index dbaa137bfb37..801d9bc3692b 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datapage/DataPageReader.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datapage/DataPageReader.java @@ -21,7 +21,7 @@ import static org.apache.ignite.internal.pagememory.util.PageIdUtils.pageId; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.io.DataPageIo; import org.apache.ignite.internal.pagememory.io.DataPagePayload; import org.jetbrains.annotations.Nullable; @@ -30,7 +30,7 @@ * Contains logic for reading values from page memory data pages (this includes handling multy-page values). */ public class DataPageReader { - private final PageMemory pageMemory; + private final PartitionPageMemory pageMemory; private final int groupId; /** @@ -39,7 +39,7 @@ public class DataPageReader { * @param pageMemory Page memory that will be used to lock and access memory. * @param groupId ID of the cache group with which the reader works (all pages must belong to this group) */ - public DataPageReader(PageMemory pageMemory, int groupId) { + public DataPageReader(PartitionPageMemory pageMemory, int groupId) { this.pageMemory = pageMemory; this.groupId = groupId; } diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datapage/NonFragmentableDataPageReader.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datapage/NonFragmentableDataPageReader.java index faa291aa8825..00921b1dc094 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datapage/NonFragmentableDataPageReader.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datapage/NonFragmentableDataPageReader.java @@ -21,7 +21,7 @@ import static org.apache.ignite.internal.pagememory.util.PageIdUtils.pageId; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.io.DataPageIo; import org.apache.ignite.internal.pagememory.io.DataPagePayload; import org.jetbrains.annotations.Nullable; @@ -30,7 +30,7 @@ * Contains logic for reading values (that cannot be fragmented, that it, occupy more than one page) from page memory data pages. */ public abstract class NonFragmentableDataPageReader { - private final PageMemory pageMemory; + private final PartitionPageMemory pageMemory; private final int groupId; /** @@ -39,7 +39,7 @@ public abstract class NonFragmentableDataPageReader { * @param pageMemory page memory that will be used to lock and access memory * @param groupId ID of the cache group with which the reader works (all pages must belong to this group) */ - public NonFragmentableDataPageReader(PageMemory pageMemory, int groupId) { + public NonFragmentableDataPageReader(PartitionPageMemory pageMemory, int groupId) { this.pageMemory = pageMemory; this.groupId = groupId; } diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datastructure/DataStructure.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datastructure/DataStructure.java index 05a0cba7c7ad..5959ec7f6fe9 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datastructure/DataStructure.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datastructure/DataStructure.java @@ -31,7 +31,7 @@ import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.pagememory.FullPageId; import org.apache.ignite.internal.pagememory.PageIdAllocator; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.io.PageIo; import org.apache.ignite.internal.pagememory.reuse.ReuseBag; import org.apache.ignite.internal.pagememory.reuse.ReuseList; @@ -41,7 +41,7 @@ import org.jetbrains.annotations.Nullable; /** - * Base class for all the data structures based on {@link PageMemory}. + * Base class for all the data structures based on {@link PartitionPageMemory}. */ public abstract class DataStructure implements ManuallyCloseable { /** For tests. */ @@ -58,7 +58,7 @@ public abstract class DataStructure implements ManuallyCloseable { protected final @Nullable String grpName; /** Page memory. */ - protected final PageMemory pageMem; + protected final PartitionPageMemory pageMem; /** Reuse list. */ protected @Nullable ReuseList reuseList; @@ -85,7 +85,7 @@ public DataStructure( int grpId, @Nullable String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, byte defaultPageFlag ) { assert !StringUtils.nullOrEmpty(structureNamePrefix); diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/freelist/FreeListImpl.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/freelist/FreeListImpl.java index a70b0c31e30b..0613e690a471 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/freelist/FreeListImpl.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/freelist/FreeListImpl.java @@ -31,7 +31,7 @@ import org.apache.ignite.internal.logger.IgniteLogger; import org.apache.ignite.internal.logger.Loggers; import org.apache.ignite.internal.pagememory.PageIdAllocator; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.Storable; import org.apache.ignite.internal.pagememory.io.DataPageIo; import org.apache.ignite.internal.pagememory.io.PageIo; @@ -330,7 +330,7 @@ public FreeListImpl( String freeListNamePrefix, int grpId, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, long metaPageId, boolean initNew, @Nullable AtomicLong pageListCacheLimit diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/freelist/PagesList.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/freelist/PagesList.java index 45eae83608de..a4cb414b79d0 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/freelist/PagesList.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/freelist/PagesList.java @@ -47,7 +47,7 @@ import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.logger.IgniteLogger; import org.apache.ignite.internal.pagememory.PageIdAllocator; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.datastructure.DataStructure; import org.apache.ignite.internal.pagememory.freelist.io.PagesListMetaIo; import org.apache.ignite.internal.pagememory.freelist.io.PagesListNodeIo; @@ -192,7 +192,7 @@ protected PagesList( String pageListNamePrefix, int grpId, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, IgniteLogger log, int buckets, long metaPageId diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemory.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemory.java index d765747a24f3..191c19a2ff18 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemory.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemory.java @@ -18,11 +18,10 @@ package org.apache.ignite.internal.pagememory.inmemory; import static java.lang.System.lineSeparator; -import static org.apache.ignite.internal.util.GridUnsafe.wrapPointer; +import static org.apache.ignite.internal.util.OffheapReadWriteLock.TAG_LOCK_ALWAYS; import java.io.Closeable; import java.io.IOException; -import java.nio.ByteBuffer; import java.util.Arrays; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -31,6 +30,7 @@ import org.apache.ignite.internal.logger.IgniteLogger; import org.apache.ignite.internal.logger.Loggers; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.configuration.VolatileDataRegionConfiguration; import org.apache.ignite.internal.pagememory.io.PageIo; import org.apache.ignite.internal.pagememory.io.PageIoRegistry; @@ -219,6 +219,11 @@ public VolatilePageMemory( started = true; } + @Override + public PartitionPageMemory createPartitionPageMemory(int groupId, int partitionId) { + return new VolatilePageMemoryDelegate(this, groupId, partitionId); + } + @Override public void stop(boolean deallocate) throws IgniteInternalException { synchronized (segmentsLock) { @@ -238,11 +243,7 @@ public void stop(boolean deallocate) throws IgniteInternalException { } } - @Override public ByteBuffer pageBuffer(long pageAddr) { - return wrapPointer(pageAddr, pageSize()); - } - - @Override public long allocatePageNoReuse(int grpId, int partId, byte flags) { + long allocatePageNoReuse(int partId, byte flags) { assert started; long relPtr = borrowFreePage(); @@ -303,7 +304,7 @@ public void stop(boolean deallocate) throws IgniteInternalException { return pageId; } - @Override public boolean freePage(int grpId, long pageId) { + boolean freePage(long pageId) { assert started; releaseFreePage(pageId); @@ -319,7 +320,7 @@ public void stop(boolean deallocate) throws IgniteInternalException { return sysPageSize; } - @Override public int realPageSize(int grpId) { + int realPageSize() { return pageSize(); } @@ -405,9 +406,7 @@ private long fromSegmentIndex(int segIdx, long pageIdx) { return res; } - // *** PageSupport methods *** - - @Override public long acquirePage(int cacheId, long pageId) { + long acquirePage(long pageId) { assert started; int pageIdx = PageIdUtils.pageIndex(pageId); @@ -417,7 +416,7 @@ private long fromSegmentIndex(int segIdx, long pageIdx) { return seg.acquirePage(pageIdx); } - @Override public void releasePage(int cacheId, long pageId, long page) { + void releasePage(long pageId) { assert started; if (trackAcquiredPages) { @@ -427,7 +426,7 @@ private long fromSegmentIndex(int segIdx, long pageIdx) { } } - @Override public long readLock(int cacheId, long pageId, long page) { + long readLock(long pageId, long page) { assert started; if (rwLock.readLock(page + LOCK_OFFSET, PageIdUtils.tag(pageId))) { @@ -437,33 +436,33 @@ private long fromSegmentIndex(int segIdx, long pageIdx) { return 0L; } - @Override public long readLockForce(int cacheId, long pageId, long page) { + long readLockForce(long page) { assert started; - if (rwLock.readLock(page + LOCK_OFFSET, -1)) { + if (rwLock.readLock(page + LOCK_OFFSET, TAG_LOCK_ALWAYS)) { return page + PAGE_OVERHEAD; } return 0L; } - @Override public void readUnlock(int cacheId, long pageId, long page) { + void readUnlock(long page) { assert started; rwLock.readUnlock(page + LOCK_OFFSET); } - @Override public long writeLock(int cacheId, long pageId, long page) { + long writeLock(long pageId, long page, boolean force) { assert started; - if (rwLock.writeLock(page + LOCK_OFFSET, PageIdUtils.tag(pageId))) { + if (rwLock.writeLock(page + LOCK_OFFSET, force ? TAG_LOCK_ALWAYS : PageIdUtils.tag(pageId))) { return page + PAGE_OVERHEAD; } return 0L; } - @Override public long tryWriteLock(int cacheId, long pageId, long page) { + long tryWriteLock(long pageId, long page) { assert started; if (rwLock.tryWriteLock(page + LOCK_OFFSET, PageIdUtils.tag(pageId))) { @@ -473,13 +472,7 @@ private long fromSegmentIndex(int segIdx, long pageIdx) { return 0L; } - @Override - public void writeUnlock( - int cacheId, - long pageId, - long page, - boolean dirtyFlag - ) { + void writeUnlock(long page) { assert started; long actualId = PageIo.getPageId(page + PAGE_OVERHEAD); @@ -487,13 +480,7 @@ public void writeUnlock( rwLock.writeUnlock(page + LOCK_OFFSET, PageIdUtils.tag(actualId)); } - @Override public boolean isDirty(int cacheId, long pageId, long page) { - // always false for page no store. - return false; - } - - @Override - public PageIoRegistry ioRegistry() { + PageIoRegistry ioRegistry() { return ioRegistry; } diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemoryDelegate.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemoryDelegate.java new file mode 100644 index 000000000000..636e7b9774d1 --- /dev/null +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemoryDelegate.java @@ -0,0 +1,114 @@ +/* + * 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.inmemory; + +import org.apache.ignite.internal.lang.IgniteInternalCheckedException; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; +import org.apache.ignite.internal.pagememory.io.PageIoRegistry; + +class VolatilePageMemoryDelegate implements PartitionPageMemory { + private final VolatilePageMemory delegate; + private final int groupId; + private final int partitionId; + + VolatilePageMemoryDelegate(VolatilePageMemory delegate, int groupId, int partitionId) { + this.delegate = delegate; + this.groupId = groupId; + this.partitionId = partitionId; + } + + @Override + public int groupId() { + return groupId; + } + + @Override + public int partitionId() { + return partitionId; + } + + @Override + public PageIoRegistry ioRegistry() { + return delegate.ioRegistry(); + } + + @Override + public int pageSize() { + return delegate.pageSize(); + } + + @Override + public int realPageSize(int groupId) { + return delegate.realPageSize(); + } + + @Override + public long allocatePageNoReuse(int groupId, int partitionId, byte flags) throws IgniteInternalCheckedException { + return delegate.allocatePageNoReuse(partitionId, flags); + } + + @Override + public boolean freePage(int groupId, long pageId) { + return delegate.freePage(pageId); + } + + @Override + public long acquirePage(int groupId, long pageId) throws IgniteInternalCheckedException { + return delegate.acquirePage(pageId); + } + + @Override + public void releasePage(int groupId, long pageId, long page) { + delegate.releasePage(pageId); + } + + @Override + public long readLock(int groupId, long pageId, long page) { + return delegate.readLock(pageId, page); + } + + @Override + public long readLockForce(int groupId, long pageId, long page) { + return delegate.readLockForce(page); + } + + @Override + public void readUnlock(int groupId, long pageId, long page) { + delegate.readUnlock(page); + } + + @Override + public long writeLock(int groupId, long pageId, long page) { + return delegate.writeLock(pageId, page, false); + } + + @Override + public long writeLockForce(int groupId, long pageId, long page) { + return delegate.writeLock(pageId, page, true); + } + + @Override + public long tryWriteLock(int groupId, long pageId, long page) { + return delegate.tryWriteLock(pageId, page); + } + + @Override + public void writeUnlock(int groupId, long pageId, long page, boolean dirtyFlag) { + delegate.writeUnlock(page); + } +} diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/io/DataPageIo.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/io/DataPageIo.java index e87d4c693f43..35e6fb13ca77 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/io/DataPageIo.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/io/DataPageIo.java @@ -18,6 +18,7 @@ package org.apache.ignite.internal.pagememory.io; import static org.apache.ignite.internal.pagememory.PageIdAllocator.FLAG_DATA; +import static org.apache.ignite.internal.util.GridUnsafe.wrapPointer; import java.nio.ByteBuffer; import java.util.Arrays; @@ -25,7 +26,7 @@ import java.util.HashSet; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.lang.IgniteStringBuilder; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.Storable; import org.apache.ignite.internal.pagememory.util.PageIdUtils; import org.apache.ignite.internal.pagememory.util.PageUtils; @@ -1124,7 +1125,7 @@ private int getDataOffsetForWrite(long pageAddr, int fullEntrySize, int directCn * @return Written payload size. */ public int addRowFragment( - PageMemory pageMem, + PartitionPageMemory pageMem, long pageId, long pageAddr, Storable row, @@ -1155,7 +1156,7 @@ public int addRowFragment( int fullEntrySize = getPageEntrySize(payloadSize, SHOW_PAYLOAD_LEN | SHOW_LINK | SHOW_ITEM); int dataOff = getDataOffsetForWrite(pageAddr, fullEntrySize, directCnt, indirectCnt, pageSize); - ByteBuffer buf = pageMem.pageBuffer(pageAddr); + ByteBuffer buf = wrapPointer(pageAddr, pageMem.pageSize()); buf.position(dataOff); diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemory.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemory.java index fc1df828f97e..085c2ef3f404 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemory.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemory.java @@ -18,6 +18,7 @@ package org.apache.ignite.internal.pagememory.persistence; import static java.lang.System.lineSeparator; +import static org.apache.ignite.internal.pagememory.PageIdAllocator.MAX_PARTITION_ID; import static org.apache.ignite.internal.pagememory.io.PageIo.getCrc; import static org.apache.ignite.internal.pagememory.io.PageIo.getPageId; import static org.apache.ignite.internal.pagememory.io.PageIo.getType; @@ -83,6 +84,7 @@ import org.apache.ignite.internal.logger.Loggers; import org.apache.ignite.internal.pagememory.FullPageId; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.configuration.PersistentDataRegionConfiguration; import org.apache.ignite.internal.pagememory.configuration.ReplacementMode; import org.apache.ignite.internal.pagememory.io.PageIoRegistry; @@ -330,6 +332,11 @@ private static List allocateRegions(long[] sizes, DirectMemo return regions; } + @Override + public PartitionPageMemory createPartitionPageMemory(int groupId, int partitionId) { + return new PersistentPageMemoryDelegate(this, groupId, partitionId); + } + /** {@inheritDoc} */ @Override public void stop(boolean deallocate) throws IgniteInternalException { @@ -350,9 +357,7 @@ public void stop(boolean deallocate) throws IgniteInternalException { } } - /** {@inheritDoc} */ - @Override - public void releasePage(int grpId, long pageId, long page) { + void releasePage(int grpId, long pageId, long page) { assert started; Segment seg = segment(grpId, pageId); @@ -366,9 +371,7 @@ public void releasePage(int grpId, long pageId, long page) { } } - /** {@inheritDoc} */ - @Override - public long readLock(int grpId, long pageId, long page) { + long readLock(long pageId, long page) { assert started; return readLock(page, pageId, false); @@ -407,17 +410,13 @@ private long readLock(long absPtr, long pageId, boolean force) { return readLock(absPtr, pageId, force, true); } - /** {@inheritDoc} */ - @Override - public void readUnlock(int grpId, long pageId, long page) { + void readUnlock(long page) { assert started; readUnlockPage(page); } - /** {@inheritDoc} */ - @Override - public long writeLock(int grpId, long pageId, long page) { + long writeLock(int grpId, long pageId, long page) { assert started; return writeLock(grpId, pageId, page, false); @@ -438,17 +437,13 @@ public long writeLock(int grpId, long pageId, long page, boolean restore) { return writeLockPage(page, new FullPageId(pageId, grpId), !restore); } - /** {@inheritDoc} */ - @Override - public long tryWriteLock(int grpId, long pageId, long page) { + long tryWriteLock(int grpId, long pageId, long page) { assert started; return tryWriteLockPage(page, new FullPageId(pageId, grpId), true); } - /** {@inheritDoc} */ - @Override - public void writeUnlock(int grpId, long pageId, long page, boolean dirtyFlag) { + void writeUnlock(int grpId, long pageId, long page, boolean dirtyFlag) { assert started; writeUnlock(grpId, pageId, page, dirtyFlag, false); @@ -469,14 +464,6 @@ public void writeUnlock(int grpId, long pageId, long page, boolean dirtyFlag, bo writeUnlockPage(page, new FullPageId(pageId, grpId), dirtyFlag, restore); } - /** {@inheritDoc} */ - @Override - public boolean isDirty(int grpId, long pageId, long page) { - assert started; - - return isDirty(page); - } - /** * Returns {@code true} if page is dirty. * @@ -486,8 +473,7 @@ boolean isDirty(long absPtr) { return dirty(absPtr); } - @Override - public long allocatePageNoReuse(int grpId, int partId, byte flags) throws IgniteInternalCheckedException { + long allocatePageNoReuse(int grpId, int partId, byte flags) throws IgniteInternalCheckedException { assert partId >= 0 && partId <= MAX_PARTITION_ID : "grpId=" + grpId + ", partId=" + partId; assert started : "grpId=" + grpId + ", partId=" + partId; @@ -580,23 +566,7 @@ public long allocatePageNoReuse(int grpId, int partId, byte flags) throws Ignite return pageId; } - /** {@inheritDoc} */ - @Override - public ByteBuffer pageBuffer(long pageAddr) { - return wrapPointer(pageAddr, pageSize()); - } - - /** {@inheritDoc} */ - @Override - public boolean freePage(int grpId, long pageId) { - assert false : "Free page should be never called directly when persistence is enabled."; - - return false; - } - - /** {@inheritDoc} */ - @Override - public long acquirePage(int grpId, long pageId) throws IgniteInternalCheckedException { + long acquirePage(int grpId, long pageId) throws IgniteInternalCheckedException { assert started : "grpId=" + grpId + ", pageId=" + hexLong(pageId); assert pageIndex(pageId) != 0 : String.format( "Partition meta should should not be read through PageMemory so as not to occupy memory: [grpId=%s, pageId=%s]", @@ -817,9 +787,8 @@ public int systemPageSize() { return sysPageSize; } - /** {@inheritDoc} */ - @Override - public int realPageSize(int grpId) { + @SuppressWarnings("PMD.UnusedFormalParameter") + int realPageSize(int grpId) { return pageSize(); } @@ -1026,9 +995,7 @@ public long acquiredPages() { return total; } - /** {@inheritDoc} */ - @Override - public long readLockForce(int grpId, long pageId, long page) { + long readLockForce(long pageId, long page) { assert started; return readLock(page, pageId, true); @@ -1749,9 +1716,7 @@ boolean shouldThrottle(double dirtyRatioThreshold) { } } - /** {@inheritDoc} */ - @Override - public PageIoRegistry ioRegistry() { + PageIoRegistry ioRegistry() { return ioRegistry; } diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemoryDelegate.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemoryDelegate.java new file mode 100644 index 000000000000..50f70e19fa70 --- /dev/null +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/PersistentPageMemoryDelegate.java @@ -0,0 +1,155 @@ +/* + * 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; + +import org.apache.ignite.internal.lang.IgniteInternalCheckedException; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; +import org.apache.ignite.internal.pagememory.io.PageIoRegistry; +import org.apache.ignite.internal.pagememory.util.PageIdUtils; + +class PersistentPageMemoryDelegate implements PartitionPageMemory { + private final PersistentPageMemory delegate; + private final int groupId; + private final int partitionId; + + PersistentPageMemoryDelegate(PersistentPageMemory delegate, int groupId, int partitionId) { + this.delegate = delegate; + this.groupId = groupId; + this.partitionId = partitionId; + } + + @Override + public int groupId() { + return groupId; + } + + @Override + public int partitionId() { + return partitionId; + } + + @Override + public PageIoRegistry ioRegistry() { + return delegate.ioRegistry(); + } + + @Override + public int pageSize() { + return delegate.pageSize(); + } + + @Override + public int realPageSize(int groupId) { + validateGroupId(groupId); + + return delegate.realPageSize(groupId); + } + + @Override + public long allocatePageNoReuse(int groupId, int partitionId, byte flags) throws IgniteInternalCheckedException { + validateGroupId(groupId); + validatePartitionId(partitionId); + + return delegate.allocatePageNoReuse(groupId, partitionId, flags); + } + + @Override + public boolean freePage(int groupId, long pageId) { + throw new UnsupportedOperationException("Free page should be never called directly when persistence is enabled."); + } + + @Override + public long acquirePage(int groupId, long pageId) throws IgniteInternalCheckedException { + validateGroupId(groupId); + validatePartitionId(PageIdUtils.partitionId(pageId)); + + return delegate.acquirePage(groupId, pageId); + } + + @Override + public void releasePage(int groupId, long pageId, long page) { + validateGroupId(groupId); + validatePartitionId(PageIdUtils.partitionId(pageId)); + + delegate.releasePage(groupId, pageId, page); + } + + @Override + public long readLock(int groupId, long pageId, long page) { + validateGroupId(groupId); + validatePartitionId(PageIdUtils.partitionId(pageId)); + + return delegate.readLock(pageId, page); + } + + @Override + public long readLockForce(int groupId, long pageId, long page) { + validateGroupId(groupId); + validatePartitionId(PageIdUtils.partitionId(pageId)); + + return delegate.readLockForce(pageId, page); + } + + @Override + public void readUnlock(int groupId, long pageId, long page) { + validateGroupId(groupId); + validatePartitionId(PageIdUtils.partitionId(pageId)); + + delegate.readUnlock(page); + } + + @Override + public long writeLock(int groupId, long pageId, long page) { + validateGroupId(groupId); + validatePartitionId(PageIdUtils.partitionId(pageId)); + + return delegate.writeLock(groupId, pageId, page); + } + + @Override + public long writeLockForce(int groupId, long pageId, long page) { + validateGroupId(groupId); + validatePartitionId(PageIdUtils.partitionId(pageId)); + + return delegate.writeLock(groupId, pageId, page, true); + } + + @Override + public long tryWriteLock(int groupId, long pageId, long page) { + validateGroupId(groupId); + validatePartitionId(PageIdUtils.partitionId(pageId)); + + return delegate.tryWriteLock(groupId, pageId, page); + } + + @Override + public void writeUnlock(int groupId, long pageId, long page, boolean dirtyFlag) { + validateGroupId(groupId); + validatePartitionId(PageIdUtils.partitionId(pageId)); + + delegate.writeUnlock(groupId, pageId, page, dirtyFlag); + } + + private void validateGroupId(int groupId) { + assert this.groupId == groupId : "Wrong groupId [expected=" + this.groupId + ", actual=" + groupId + ']'; + } + + private void validatePartitionId(int partitionId) { + assert this.partitionId == partitionId : "Wrong partitionId [expected=" + this.partitionId + ", actual=" + partitionId + ']'; + } +} diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/tree/BplusTree.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/tree/BplusTree.java index f6eb1bf101b2..84ae7424cbd1 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/tree/BplusTree.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/tree/BplusTree.java @@ -54,7 +54,7 @@ import org.apache.ignite.internal.lang.IgniteStringBuilder; import org.apache.ignite.internal.lang.IgniteTuple3; import org.apache.ignite.internal.pagememory.CorruptedDataStructureException; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.datastructure.DataStructure; import org.apache.ignite.internal.pagememory.io.IoVersions; import org.apache.ignite.internal.pagememory.io.PageIo; @@ -924,7 +924,7 @@ protected BplusTree( int grpId, @Nullable String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList, @@ -954,7 +954,7 @@ protected BplusTree( int grpId, @Nullable String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/util/PageHandler.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/util/PageHandler.java index dba6d06464c6..0802b5ee399a 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/util/PageHandler.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/util/PageHandler.java @@ -20,7 +20,7 @@ import static java.lang.Boolean.FALSE; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.io.PageIo; import org.jetbrains.annotations.Nullable; @@ -93,7 +93,7 @@ default boolean releaseAfterWrite( * @throws IgniteInternalCheckedException If failed. */ static R readPage( - PageMemory pageMem, + PartitionPageMemory pageMem, int groupId, long pageId, PageHandler h, @@ -125,7 +125,7 @@ static R readPage( * @throws IgniteInternalCheckedException If failed. */ static R readPage( - PageMemory pageMem, + PartitionPageMemory pageMem, int groupId, long pageId, long page, @@ -162,7 +162,7 @@ static R readPage( * @see PageIo#initNewPage(long, long, int) */ static void initPage( - PageMemory pageMem, + PartitionPageMemory pageMem, int groupId, long pageId, PageIo init @@ -197,7 +197,7 @@ static void initPage( */ // TODO IGNITE-16350 Consider splitting into two separate methods for init and regular locking. static R writePage( - PageMemory pageMem, + PartitionPageMemory pageMem, int groupId, final long pageId, PageHandler h, @@ -260,7 +260,7 @@ static R writePage( * @throws IgniteInternalCheckedException If failed. */ static R writePage( - PageMemory pageMem, + PartitionPageMemory pageMem, int groupId, long pageId, long page, @@ -304,7 +304,7 @@ static R writePage( * Invokes {@link PageIo#initNewPage(long, long, int)} and does additional checks. */ private static void doInitPage( - PageMemory pageMem, + PartitionPageMemory pageMem, int groupId, long pageId, long pageAddr, diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/freelist/FreeListImplTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/freelist/FreeListImplTest.java index d5e33bf233db..da7d7ac2e6d1 100644 --- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/freelist/FreeListImplTest.java +++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/freelist/FreeListImplTest.java @@ -39,6 +39,7 @@ import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.Storable; import org.apache.ignite.internal.pagememory.TestPageIoRegistry; import org.apache.ignite.internal.pagememory.configuration.VolatileDataRegionConfiguration; @@ -127,14 +128,15 @@ private static Stream provideTestArguments() { private FreeList createFreeList(int pageSize) throws Exception { pageMemory = createPageMemory(pageSize); + PartitionPageMemory partitionPageMemory = pageMemory.createPartitionPageMemory(1, 1); - long metaPageId = pageMemory.allocatePageNoReuse(1, 1, FLAG_DATA); + long metaPageId = partitionPageMemory.allocatePageNoReuse(1, 1, FLAG_DATA); return new FreeListImpl( "TestFreeList", 0, 1, - pageMemory, + partitionPageMemory, metaPageId, true, null diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemoryNoLoadSelfTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemoryNoLoadSelfTest.java index 653a6145d924..5b75fbd6202a 100644 --- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemoryNoLoadSelfTest.java +++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/inmemory/VolatilePageMemoryNoLoadSelfTest.java @@ -22,6 +22,7 @@ import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension; import org.apache.ignite.internal.pagememory.AbstractPageMemoryNoLoadSelfTest; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.configuration.VolatileDataRegionConfiguration; import org.apache.ignite.internal.pagememory.io.PageIoRegistry; import org.apache.ignite.internal.pagememory.mem.IgniteOutOfMemoryException; @@ -51,13 +52,14 @@ protected VolatilePageMemory memory() { @Test public void testLoadedPagesCount() { VolatilePageMemory mem = memory(); + PartitionPageMemory partitionPageMemory = mem.createPartitionPageMemory(GRP_ID, PARTITION_ID); int expPages = MAX_MEMORY_SIZE / mem.systemPageSize(); try { assertThrows(IgniteOutOfMemoryException.class, () -> { for (int i = 0; i < expPages * 2; i++) { - allocatePage(mem); + allocatePage(partitionPageMemory); } }); diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/replacement/AbstractPageReplacementTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/replacement/AbstractPageReplacementTest.java index 765251740665..0f54818f4347 100644 --- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/replacement/AbstractPageReplacementTest.java +++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/replacement/AbstractPageReplacementTest.java @@ -63,6 +63,7 @@ import org.apache.ignite.internal.lang.RunnableX; import org.apache.ignite.internal.metrics.MetricSet; import org.apache.ignite.internal.pagememory.DataRegion; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.TestDataRegion; import org.apache.ignite.internal.pagememory.TestPageIoModule.TestSimpleValuePageIo; import org.apache.ignite.internal.pagememory.TestPageIoRegistry; @@ -118,6 +119,7 @@ public abstract class AbstractPageReplacementTest extends IgniteAbstractTest { private CheckpointManager checkpointManager; private PersistentPageMemory pageMemory; + private PartitionPageMemory partitionPageMemory; @InjectExecutorService private ExecutorService executorService; @@ -178,6 +180,8 @@ void setUp() throws Exception { checkpointManager.partitionDestructionLockManager() ); + partitionPageMemory = pageMemory.createPartitionPageMemory(GROUP_ID, PARTITION_ID); + dataRegionList.add(new TestDataRegion<>(pageMemory)); filePageStoreManager.start(); @@ -378,20 +382,20 @@ private void assertMetricValue(String metricName, Matcher valueMatcher) { } private void createAndFillTestSimpleValuePage(long pageId) throws Exception { - long page = pageMemory.acquirePage(GROUP_ID, pageId); + long page = partitionPageMemory.acquirePage(GROUP_ID, pageId); try { - long pageAddr = pageMemory.writeLock(GROUP_ID, pageId, page); + long pageAddr = partitionPageMemory.writeLock(GROUP_ID, pageId, page); try { new TestSimpleValuePageIo().initNewPage(pageAddr, pageId, PAGE_SIZE); TestSimpleValuePageIo.setLongValue(pageAddr, pageIndex(pageId) * 3L); } finally { - pageMemory.writeUnlock(GROUP_ID, pageId, page, true); + partitionPageMemory.writeUnlock(GROUP_ID, pageId, page, true); } } finally { - pageMemory.releasePage(GROUP_ID, pageId, page); + partitionPageMemory.releasePage(GROUP_ID, pageId, page); } } @@ -454,13 +458,13 @@ private void inCheckpointReadLock(RunnableX runnableX) throws Throwable { private void createAndFillTestSimpleValuePages(int pageCount) throws Exception { for (int i = 0; i < pageCount; i++) { - createAndFillTestSimpleValuePage(pageMemory.allocatePage(null, GROUP_ID, PARTITION_ID, FLAG_DATA)); + createAndFillTestSimpleValuePage(partitionPageMemory.allocatePage(null, GROUP_ID, PARTITION_ID, FLAG_DATA)); } } private void createAndFillTestSimpleValuePages(BooleanSupplier continuePredicate) throws Exception { while (continuePredicate.getAsBoolean()) { - createAndFillTestSimpleValuePage(pageMemory.allocatePage(null, GROUP_ID, PARTITION_ID, FLAG_DATA)); + createAndFillTestSimpleValuePage(partitionPageMemory.allocatePage(null, GROUP_ID, PARTITION_ID, FLAG_DATA)); } } } diff --git a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/throttling/PageMemoryThrottlingTest.java b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/throttling/PageMemoryThrottlingTest.java index 60db4a7de08d..1a37f083de76 100644 --- a/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/throttling/PageMemoryThrottlingTest.java +++ b/modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/throttling/PageMemoryThrottlingTest.java @@ -53,6 +53,7 @@ import org.apache.ignite.internal.lang.RunnableX; import org.apache.ignite.internal.pagememory.DataRegion; import org.apache.ignite.internal.pagememory.PageIdAllocator; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.TestDataRegion; import org.apache.ignite.internal.pagememory.configuration.CheckpointConfiguration; import org.apache.ignite.internal.pagememory.configuration.PersistentDataRegionConfiguration; @@ -114,6 +115,8 @@ public class PageMemoryThrottlingTest extends IgniteAbstractTest { private PersistentPageMemory pageMemory; + private PartitionPageMemory partitionPageMemory; + private FileIoFactory fileIoFactory; private DataRegion dataRegion; @@ -186,6 +189,8 @@ private void setUp(ThrottlingPolicyFactory throttleFactory) throws Exception { checkpointManager.partitionDestructionLockManager() ); + partitionPageMemory = pageMemory.createPartitionPageMemory(GROUP_ID, PART_ID); + pageStoreManager.start(); dataRegion = new TestDataRegion<>(pageMemory); @@ -233,7 +238,7 @@ void pageAllocationNotifiedThrottler() throws Exception { checkpointManager.checkpointTimeoutLock().checkpointReadLock(); try { - pageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX); + partitionPageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX); verify(writeThrottle).onMarkDirty(eq(false)); } finally { @@ -254,7 +259,7 @@ void pageUnlockWithoutMarkingDirty() throws Exception { checkpointManager.checkpointTimeoutLock().checkpointReadLock(); try { - pageId = pageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX); + pageId = partitionPageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX); initPage(pageId); } finally { @@ -288,7 +293,7 @@ void pageMarkedDirtyOnlyOnce() throws Exception { AtomicLong pageId = new AtomicLong(); runInLock(() -> { - pageId.set(pageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX)); + pageId.set(partitionPageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX)); initPage(pageId.get()); }); @@ -318,7 +323,7 @@ void checkpointEvents() throws Exception { AtomicLong pageId = new AtomicLong(); runInLock(() -> { - pageId.set(pageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX)); + pageId.set(partitionPageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX)); initPage(pageId.get()); }); @@ -346,7 +351,7 @@ void wakeupThrottledThreads() throws Exception { runInLock(() -> { for (int i = 0; i < pages; i++) { - long pageId = pageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX); + long pageId = partitionPageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX); pageIds[i] = pageId; @@ -413,7 +418,7 @@ void hugeLoadDoesNotBreakCheckpointReadLock(boolean speedBasedThrottling) throws // TODO https://issues.apache.org/jira/browse/IGNITE-24877 This line should not be necessary. checkpointManager.markPartitionAsDirty(dataRegion, GROUP_ID, PART_ID, 1); - long pageId = pageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX); + long pageId = partitionPageMemory.allocatePageNoReuse(GROUP_ID, PART_ID, PageIdAllocator.FLAG_AUX); pageIds[i] = pageId; @@ -447,34 +452,34 @@ private void runInLock(RunnableX runnable) { } private void initPage(long pageId) throws IgniteInternalCheckedException { - long page = pageMemory.acquirePage(GROUP_ID, pageId); + long page = partitionPageMemory.acquirePage(GROUP_ID, pageId); try { - long pageAddr = pageMemory.writeLock(GROUP_ID, pageId, page, true); + long pageAddr = partitionPageMemory.writeLockForce(GROUP_ID, pageId, page); try { PagesListNodeIo.VERSIONS.latest().initNewPage(pageAddr, pageId, PAGE_SIZE); } finally { - pageMemory.writeUnlock(GROUP_ID, pageId, page, true); + partitionPageMemory.writeUnlock(GROUP_ID, pageId, page, true); } } finally { - pageMemory.releasePage(GROUP_ID, pageId, page); + partitionPageMemory.releasePage(GROUP_ID, pageId, page); } } private void acquireAndReleaseWriteLock(long pageId, boolean markDirty) throws IgniteInternalCheckedException { - long page = pageMemory.acquirePage(GROUP_ID, pageId); + long page = partitionPageMemory.acquirePage(GROUP_ID, pageId); try { - long pageAddr = pageMemory.writeLock(GROUP_ID, pageId, page); + long pageAddr = partitionPageMemory.writeLock(GROUP_ID, pageId, page); try { assertNotEquals(0L, pageAddr); } finally { - pageMemory.writeUnlock(GROUP_ID, pageId, page, markDirty); + partitionPageMemory.writeUnlock(GROUP_ID, pageId, page, markDirty); } } finally { - pageMemory.releasePage(GROUP_ID, pageId, page); + partitionPageMemory.releasePage(GROUP_ID, pageId, page); } } } diff --git a/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/AbstractPageMemoryNoLoadSelfTest.java b/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/AbstractPageMemoryNoLoadSelfTest.java index 448c5ec87096..d4306e611813 100644 --- a/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/AbstractPageMemoryNoLoadSelfTest.java +++ b/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/AbstractPageMemoryNoLoadSelfTest.java @@ -53,7 +53,8 @@ public abstract class AbstractPageMemoryNoLoadSelfTest extends BaseIgniteAbstrac @Test public void testPageTearingInner() throws Exception { - PageMemory mem = memory(); + PageMemory pageMemory = memory(); + PartitionPageMemory mem = pageMemory.createPartitionPageMemory(GRP_ID, PARTITION_ID); try { FullPageId fullId1 = allocatePage(mem); @@ -84,13 +85,14 @@ public void testPageTearingInner() throws Exception { mem.releasePage(fullId1.groupId(), fullId1.pageId(), page1); } } finally { - mem.stop(true); + pageMemory.stop(true); } } @Test public void testPageTearingSequential() throws Exception { - PageMemory mem = memory(); + PageMemory pageMemory = memory(); + PartitionPageMemory mem = pageMemory.createPartitionPageMemory(GRP_ID, PARTITION_ID); try { int pagesCnt = 1024; @@ -131,13 +133,14 @@ public void testPageTearingSequential() throws Exception { } } } finally { - mem.stop(true); + pageMemory.stop(true); } } @Test public void testPageHandleDeallocation() throws Exception { - PageMemory mem = memory(); + PageMemory pageMemory = memory(); + PartitionPageMemory mem = pageMemory.createPartitionPageMemory(GRP_ID, PARTITION_ID); try { int pages = 3 * 1024 * 1024 / (8 * 1024); @@ -156,13 +159,14 @@ public void testPageHandleDeallocation() throws Exception { assertFalse(handles.add(allocatePage(mem))); } } finally { - mem.stop(true); + pageMemory.stop(true); } } @Test public void testPageIdRotation() throws Exception { - PageMemory mem = memory(); + PageMemory pageMemory = memory(); + PartitionPageMemory mem = pageMemory.createPartitionPageMemory(GRP_ID, PARTITION_ID); try { int pages = 5; @@ -250,7 +254,7 @@ public void testPageIdRotation() throws Exception { } } } finally { - mem.stop(true); + pageMemory.stop(true); } } @@ -267,7 +271,7 @@ public void testPageIdRotation() throws Exception { * @param page Page pointer. * @param val Value to write. */ - protected void writePage(PageMemory mem, FullPageId fullId, long page, int val) { + protected void writePage(PartitionPageMemory mem, FullPageId fullId, long page, int val) { long pageAddr = mem.writeLock(GRP_ID, fullId.pageId(), page); try { @@ -289,7 +293,7 @@ protected void writePage(PageMemory mem, FullPageId fullId, long page, int val) * @param page Page pointer. * @param expVal Expected value. */ - private void readPage(PageMemory mem, long pageId, long page, int expVal) { + private void readPage(PartitionPageMemory mem, long pageId, long page, int expVal) { expVal &= 0xFF; long pageAddr = mem.readLock(GRP_ID, pageId, page); diff --git a/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/benchmark/PersistentPageMemoryBenchmarkBase.java b/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/benchmark/PersistentPageMemoryBenchmarkBase.java index e88286e67f8c..2c5d16bf0049 100644 --- a/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/benchmark/PersistentPageMemoryBenchmarkBase.java +++ b/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/benchmark/PersistentPageMemoryBenchmarkBase.java @@ -24,6 +24,8 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.apache.ignite.internal.components.NoOpLogSyncer; @@ -31,6 +33,7 @@ import org.apache.ignite.internal.failure.NoOpFailureManager; import org.apache.ignite.internal.fileio.RandomAccessFileIoFactory; import org.apache.ignite.internal.pagememory.DataRegion; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.TestDataRegion; import org.apache.ignite.internal.pagememory.TestPageIoRegistry; import org.apache.ignite.internal.pagememory.configuration.CheckpointConfiguration; @@ -60,6 +63,7 @@ public class PersistentPageMemoryBenchmarkBase { private static final String NODE_NAME = "benchmark-node"; private PersistentPageMemory persistentPageMemory; + private Map partitionPageMemoryMap = new ConcurrentHashMap<>(); private CheckpointManager checkpointManager; private FilePageStoreManager filePageStoreManager; private PartitionMetaManager partitionMetaManager; @@ -71,6 +75,10 @@ protected PersistentPageMemory persistentPageMemory() { return persistentPageMemory; } + protected PartitionPageMemory partitionPageMemory(int partitionId) { + return partitionPageMemoryMap.get(new GroupPartitionId(GROUP_ID, partitionId)); + } + protected CheckpointManager checkpointManager() { return checkpointManager; } @@ -161,6 +169,10 @@ private void createPartitionFilePageStore(int partitionId, int pageSize) throws try { var groupPartitionId = new GroupPartitionId(GROUP_ID, partitionId); + assert !partitionPageMemoryMap.containsKey(groupPartitionId) : partitionPageMemoryMap; + + partitionPageMemoryMap.put(groupPartitionId, persistentPageMemory.createPartitionPageMemory(GROUP_ID, partitionId)); + FilePageStore filePageStore = filePageStoreManager.readOrCreateStore(groupPartitionId, buffer.rewind()); filePageStore.ensure(); diff --git a/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/benchmark/VolatilePageMemoryBenchmarkBase.java b/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/benchmark/VolatilePageMemoryBenchmarkBase.java index ccc85765d102..3d100d2ac5c1 100644 --- a/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/benchmark/VolatilePageMemoryBenchmarkBase.java +++ b/modules/page-memory/src/testFixtures/java/org/apache/ignite/internal/pagememory/benchmark/VolatilePageMemoryBenchmarkBase.java @@ -21,6 +21,7 @@ import java.util.Random; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.configuration.VolatileDataRegionConfiguration; import org.apache.ignite.internal.pagememory.freelist.FreeList; import org.apache.ignite.internal.pagememory.freelist.FreeListImpl; @@ -68,12 +69,14 @@ public void setup() throws Exception { new OffheapReadWriteLock(OffheapReadWriteLock.DEFAULT_CONCURRENCY_LEVEL) ); + PartitionPageMemory partitionPageMemory = volatilePageMemory.createPartitionPageMemory(GROUP_ID, PARTITION_ID); + freeList = new FreeListImpl( "freeList", GROUP_ID, PARTITION_ID, - volatilePageMemory, - volatilePageMemory.allocatePageNoReuse(GROUP_ID, PARTITION_ID, FLAG_AUX), + partitionPageMemory, + partitionPageMemory.allocatePageNoReuse(GROUP_ID, PARTITION_ID, FLAG_AUX), true, null ); diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java index 426aecca7918..d32d97c16f16 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java @@ -34,6 +34,7 @@ import org.apache.ignite.internal.lang.IgniteInternalException; import org.apache.ignite.internal.lang.IgniteStringFormatter; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.freelist.FreeListImpl; import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId; import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory; @@ -134,17 +135,20 @@ public PersistentPageMemoryMvPartitionStorage createMvPartitionStorage(int parti return inCheckpointLock(() -> { PersistentPageMemory pageMemory = dataRegion.pageMemory(); - FreeListImpl freeList = createFreeList(partitionId, pageMemory, meta); + PartitionPageMemory partitionPageMemory = pageMemory.createPartitionPageMemory(getTableId(), partitionId); - VersionChainTree versionChainTree = createVersionChainTree(partitionId, freeList, pageMemory, meta); + FreeListImpl freeList = createFreeList(partitionId, partitionPageMemory, meta); - IndexMetaTree indexMetaTree = createIndexMetaTree(partitionId, freeList, pageMemory, meta); + VersionChainTree versionChainTree = createVersionChainTree(partitionId, freeList, partitionPageMemory, meta); - GcQueue gcQueue = createGcQueue(partitionId, freeList, pageMemory, meta); + IndexMetaTree indexMetaTree = createIndexMetaTree(partitionId, freeList, partitionPageMemory, meta); + + GcQueue gcQueue = createGcQueue(partitionId, freeList, partitionPageMemory, meta); return new PersistentPageMemoryMvPartitionStorage( this, partitionId, + partitionPageMemory, meta, freeList, versionChainTree, @@ -176,7 +180,7 @@ public PersistentPageMemoryMvPartitionStorage createMvPartitionStorage(int parti */ private FreeListImpl createFreeList( int partId, - PersistentPageMemory pageMemory, + PartitionPageMemory pageMemory, StoragePartitionMeta meta ) throws StorageException { try { @@ -194,7 +198,7 @@ private FreeListImpl createFreeList( "PersistentFreeList", getTableId(), partId, - dataRegion.pageMemory(), + pageMemory, meta.freeListRootPageId(), initNew, dataRegion.pageListCacheLimit() @@ -216,7 +220,7 @@ private FreeListImpl createFreeList( private VersionChainTree createVersionChainTree( int partId, ReuseList reuseList, - PersistentPageMemory pageMemory, + PartitionPageMemory pageMemory, StoragePartitionMeta meta ) throws StorageException { try { @@ -234,7 +238,7 @@ private VersionChainTree createVersionChainTree( getTableId(), Integer.toString(getTableId()), partId, - dataRegion.pageMemory(), + pageMemory, engine.generateGlobalRemoveId(), meta.versionChainTreeRootPageId(), reuseList, @@ -257,7 +261,7 @@ private VersionChainTree createVersionChainTree( private IndexMetaTree createIndexMetaTree( int partitionId, ReuseList reuseList, - PersistentPageMemory pageMemory, + PartitionPageMemory pageMemory, StoragePartitionMeta meta ) { try { @@ -275,7 +279,7 @@ private IndexMetaTree createIndexMetaTree( getTableId(), Integer.toString(getTableId()), partitionId, - dataRegion.pageMemory(), + pageMemory, engine.generateGlobalRemoveId(), meta.indexTreeMetaPageId(), reuseList, @@ -298,7 +302,7 @@ private IndexMetaTree createIndexMetaTree( private GcQueue createGcQueue( int partitionId, ReuseList reuseList, - PersistentPageMemory pageMemory, + PartitionPageMemory pageMemory, StoragePartitionMeta meta ) { try { @@ -316,7 +320,7 @@ private GcQueue createGcQueue( getTableId(), Integer.toString(getTableId()), partitionId, - dataRegion.pageMemory(), + pageMemory, engine.generateGlobalRemoveId(), meta.gcQueueMetaPageId(), reuseList, @@ -347,22 +351,25 @@ CompletableFuture clearStorageAndUpdateDataStructures( GroupPartitionId groupPartitionId = createGroupPartitionId(mvPartitionStorage.partitionId()); return destroyPartitionPhysically(groupPartitionId).thenAccept(unused -> { - PersistentPageMemory pageMemory = dataRegion.pageMemory(); - int partitionId = groupPartitionId.getPartitionId(); StoragePartitionMeta meta = getOrCreatePartitionMetaOnCreatePartition(groupPartitionId); inCheckpointLock(() -> { - FreeListImpl freeList = createFreeList(partitionId, pageMemory, meta); + PersistentPageMemory pageMemory = dataRegion.pageMemory(); + + PartitionPageMemory partitionPageMemory = pageMemory.createPartitionPageMemory(getTableId(), partitionId); + + FreeListImpl freeList = createFreeList(partitionId, partitionPageMemory, meta); - VersionChainTree versionChainTree = createVersionChainTree(partitionId, freeList, pageMemory, meta); + VersionChainTree versionChainTree = createVersionChainTree(partitionId, freeList, partitionPageMemory, meta); - IndexMetaTree indexMetaTree = createIndexMetaTree(partitionId, freeList, pageMemory, meta); + IndexMetaTree indexMetaTree = createIndexMetaTree(partitionId, freeList, partitionPageMemory, meta); - GcQueue gcQueue = createGcQueue(partitionId, freeList, pageMemory, meta); + GcQueue gcQueue = createGcQueue(partitionId, freeList, partitionPageMemory, meta); ((PersistentPageMemoryMvPartitionStorage) mvPartitionStorage).updateDataStructures( + partitionPageMemory, meta, freeList, versionChainTree, diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryDataRegion.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryDataRegion.java index 62d75d8859ad..346c165634be 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryDataRegion.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryDataRegion.java @@ -26,7 +26,7 @@ import org.apache.ignite.internal.logger.IgniteLogger; import org.apache.ignite.internal.logger.Loggers; import org.apache.ignite.internal.pagememory.DataRegion; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.configuration.VolatileDataRegionConfiguration; import org.apache.ignite.internal.pagememory.freelist.FreeListImpl; import org.apache.ignite.internal.pagememory.inmemory.VolatilePageMemory; @@ -98,7 +98,7 @@ public void start() { var pageMemory = new VolatilePageMemory(regionConfiguration, ioRegistry, new OffheapReadWriteLock(lockConcLvl)); try { - this.freeList = createFreeList(pageMemory); + this.freeList = createFreeList(pageMemory.createPartitionPageMemory(FREE_LIST_GROUP_ID, FREE_LIST_PARTITION_ID)); } catch (IgniteInternalCheckedException e) { throw new StorageException("Error creating free list", e); } @@ -138,7 +138,7 @@ private static VolatileDataRegionConfiguration regionConfiguration(VolatilePageM .build(); } - private static FreeListImpl createFreeList(PageMemory pageMemory) throws IgniteInternalCheckedException { + private static FreeListImpl createFreeList(PartitionPageMemory pageMemory) throws IgniteInternalCheckedException { long metaPageId = pageMemory.allocatePageNoReuse(FREE_LIST_GROUP_ID, FREE_LIST_PARTITION_ID, FLAG_AUX); return new FreeListImpl( diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryTableStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryTableStorage.java index 1cf367f0f5c0..e18e16354a43 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryTableStorage.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryTableStorage.java @@ -25,6 +25,7 @@ import org.apache.ignite.internal.failure.FailureContext; import org.apache.ignite.internal.failure.FailureProcessor; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.storage.StorageException; import org.apache.ignite.internal.storage.engine.StorageTableDescriptor; import org.apache.ignite.internal.storage.index.StorageIndexDescriptorSupplier; @@ -84,15 +85,17 @@ public VolatilePageMemoryStorageEngine engine() { @Override public VolatilePageMemoryMvPartitionStorage createMvPartitionStorage(int partitionId) throws StorageException { - VersionChainTree versionChainTree = createVersionChainTree(partitionId); + PartitionPageMemory partitionPageMemory = dataRegion.pageMemory().createPartitionPageMemory(getTableId(), partitionId); - IndexMetaTree indexMetaTree = createIndexMetaTree(partitionId); + VersionChainTree versionChainTree = createVersionChainTree(partitionPageMemory); - GcQueue gcQueue = createGarbageCollectionTree(partitionId); + IndexMetaTree indexMetaTree = createIndexMetaTree(partitionPageMemory); + + GcQueue gcQueue = createGarbageCollectionTree(partitionPageMemory); return new VolatilePageMemoryMvPartitionStorage( this, - partitionId, + partitionPageMemory, versionChainTree, indexMetaTree, gcQueue, @@ -101,15 +104,16 @@ public VolatilePageMemoryMvPartitionStorage createMvPartitionStorage(int partiti ); } - private IndexMetaTree createIndexMetaTree(int partitionId) { + private IndexMetaTree createIndexMetaTree(PartitionPageMemory partitionPageMemory) { try { - long metaPageId = dataRegion.pageMemory().allocatePage(dataRegion.reuseList(), getTableId(), partitionId, FLAG_AUX); + int partitionId = partitionPageMemory.partitionId(); + long metaPageId = partitionPageMemory.allocatePage(dataRegion.reuseList(), getTableId(), partitionId, FLAG_AUX); return new IndexMetaTree( getTableId(), Integer.toString(getTableId()), partitionId, - dataRegion.pageMemory(), + partitionPageMemory, engine.generateGlobalRemoveId(), metaPageId, dataRegion.reuseList(), @@ -120,15 +124,16 @@ private IndexMetaTree createIndexMetaTree(int partitionId) { } } - private GcQueue createGarbageCollectionTree(int partitionId) { + private GcQueue createGarbageCollectionTree(PartitionPageMemory partitionPageMemory) { try { - long metaPageId = dataRegion.pageMemory().allocatePage(dataRegion().reuseList(), getTableId(), partitionId, FLAG_AUX); + int partitionId = partitionPageMemory.partitionId(); + long metaPageId = partitionPageMemory.allocatePage(dataRegion().reuseList(), getTableId(), partitionId, FLAG_AUX); return new GcQueue( getTableId(), Integer.toString(getTableId()), partitionId, - dataRegion.pageMemory(), + partitionPageMemory, engine.generateGlobalRemoveId(), metaPageId, dataRegion.reuseList(), @@ -152,25 +157,26 @@ protected void finishDestruction() { /** * Returns new {@link VersionChainTree} instance for partition. * - * @param partId Partition ID. + * @param partitionPageMemory Page memory instance associated with a given partition. * @throws StorageException If failed. */ - private VersionChainTree createVersionChainTree(int partId) throws StorageException { + private VersionChainTree createVersionChainTree(PartitionPageMemory partitionPageMemory) throws StorageException { + int partitionId = partitionPageMemory.partitionId(); try { - long metaPageId = dataRegion.pageMemory().allocatePage(dataRegion().reuseList(), getTableId(), partId, FLAG_AUX); + long metaPageId = partitionPageMemory.allocatePage(dataRegion().reuseList(), getTableId(), partitionId, FLAG_AUX); return new VersionChainTree( getTableId(), Integer.toString(getTableId()), - partId, - dataRegion.pageMemory(), + partitionId, + partitionPageMemory, engine.generateGlobalRemoveId(), metaPageId, dataRegion.reuseList(), true ); } catch (IgniteInternalCheckedException e) { - throw new StorageException("Error creating TableTree: [tableId={}, partitionId={}]", e, getTableId(), partId); + throw new StorageException("Error creating TableTree: [tableId={}, partitionId={}]", e, getTableId(), partitionId); } } @@ -192,11 +198,13 @@ CompletableFuture clearStorageAndUpdateDataStructures( }); int partitionId = mvPartitionStorage.partitionId(); + PartitionPageMemory partitionPageMemory = dataRegion.pageMemory().createPartitionPageMemory(getTableId(), partitionId); volatilePartitionStorage.updateDataStructures( - createVersionChainTree(partitionId), - createIndexMetaTree(partitionId), - createGarbageCollectionTree(partitionId) + partitionPageMemory, + createVersionChainTree(partitionPageMemory), + createIndexMetaTree(partitionPageMemory), + createGarbageCollectionTree(partitionPageMemory) ); afterUpdateStructuresCallback.run(); diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/HashIndexTree.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/HashIndexTree.java index 8c21bd479b13..b3a044fdf1e0 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/HashIndexTree.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/HashIndexTree.java @@ -22,7 +22,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.datapage.DataPageReader; import org.apache.ignite.internal.pagememory.reuse.ReuseList; import org.apache.ignite.internal.pagememory.tree.BplusTree; @@ -61,7 +61,7 @@ private HashIndexTree( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList @@ -92,7 +92,7 @@ private HashIndexTree( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList, @@ -125,7 +125,7 @@ public static HashIndexTree createNew( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList, @@ -141,7 +141,7 @@ public static HashIndexTree restoreExisting( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaTree.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaTree.java index 63f9e018dcf3..945e0b69a337 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaTree.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaTree.java @@ -19,7 +19,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.reuse.ReuseList; import org.apache.ignite.internal.pagememory.tree.BplusTree; import org.apache.ignite.internal.pagememory.tree.io.BplusIo; @@ -52,7 +52,7 @@ public IndexMetaTree( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList, diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/SortedIndexTree.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/SortedIndexTree.java index 3c6087d6e336..7fbef66e9034 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/SortedIndexTree.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/SortedIndexTree.java @@ -23,7 +23,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.ignite.internal.binarytuple.BinaryTuple; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.datapage.DataPageReader; import org.apache.ignite.internal.pagememory.reuse.ReuseList; import org.apache.ignite.internal.pagememory.tree.BplusTree; @@ -85,7 +85,7 @@ private SortedIndexTree( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList, @@ -123,7 +123,7 @@ private SortedIndexTree( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList @@ -146,7 +146,7 @@ public static SortedIndexTree createNew( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList, @@ -165,7 +165,7 @@ public static SortedIndexTree restoreExisting( int grpId, @Nullable String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList, @@ -184,7 +184,7 @@ public static SortedIndexTree restoreForDestroy( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java index c4b211941e53..52324402ab59 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java @@ -37,8 +37,7 @@ import org.apache.ignite.internal.hlc.HybridTimestamp; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.lang.IgniteStringFormatter; -import org.apache.ignite.internal.pagememory.PageMemory; -import org.apache.ignite.internal.pagememory.datapage.DataPageReader; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.freelist.FreeListImpl; import org.apache.ignite.internal.pagememory.tree.BplusTree.TreeRowMapClosure; import org.apache.ignite.internal.pagememory.tree.IgniteTree.InvokeClosure; @@ -97,7 +96,7 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio protected final int partitionId; - protected final AbstractPageMemoryTableStorage tableStorage; + protected final AbstractPageMemoryTableStorage tableStorage; final PageMemoryIndexes indexes; @@ -113,8 +112,6 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio volatile RenewablePartitionStorageState renewableState; - protected final DataPageReader rowVersionDataPageReader; - /** Busy lock. */ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock(); @@ -126,7 +123,7 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio */ AbstractPageMemoryMvPartitionStorage( int partitionId, - AbstractPageMemoryTableStorage tableStorage, + AbstractPageMemoryTableStorage tableStorage, RenewablePartitionStorageState renewableState, ExecutorService destructionExecutor, FailureProcessor failureProcessor @@ -137,10 +134,6 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio this.destructionExecutor = createGradualTaskExecutor(destructionExecutor); this.failureProcessor = failureProcessor; this.indexes = new PageMemoryIndexes(this.destructionExecutor, failureProcessor, this::runConsistently); - - PageMemory pageMemory = tableStorage.dataRegion().pageMemory(); - - rowVersionDataPageReader = new DataPageReader(pageMemory, tableStorage.getTableId()); } protected abstract GradualTaskExecutor createGradualTaskExecutor(ExecutorService threadPool); @@ -194,6 +187,7 @@ public void createSortedIndex(StorageSortedIndexDescriptor indexDescriptor) { } void updateRenewableState( + PartitionPageMemory partitionPageMemory, VersionChainTree versionChainTree, FreeListImpl freeList, IndexMetaTree indexMetaTree, @@ -201,7 +195,7 @@ void updateRenewableState( ) { var newState = new RenewablePartitionStorageState( tableStorage, - partitionId, + partitionPageMemory, versionChainTree, freeList, indexMetaTree, @@ -277,7 +271,7 @@ RowVersion readRowVersion(long rowVersionLink, Predicate loadVa ReadRowVersion read = new ReadRowVersion(partitionId); try { - rowVersionDataPageReader.traverse(rowVersionLink, read, loadValue); + renewableState.rowVersionDataPageReader().traverse(rowVersionLink, read, loadValue); } catch (IgniteInternalCheckedException e) { throw new StorageException("Row version lookup failed: [link={}, {}]", e, rowVersionLink, createStorageInfo()); } @@ -291,7 +285,7 @@ RowVersion readRowVersion(long rowVersionLink, Predicate loadVa FindRowVersion findRowVersion = new FindRowVersion(partitionId, loadValueBytes); try { - rowVersionDataPageReader.traverse(versionChain.headLink(), findRowVersion, filter); + renewableState.rowVersionDataPageReader().traverse(versionChain.headLink(), findRowVersion, filter); } catch (IgniteInternalCheckedException e) { throw new StorageException( "Error when looking up row version in version chain: [rowId={}, headLink={}, {}]", diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/BlobStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/BlobStorage.java index d47a86b015a7..bd2814baf9b9 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/BlobStorage.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/BlobStorage.java @@ -20,7 +20,7 @@ import java.util.Objects; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.pagememory.PageIdAllocator; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.datastructure.DataStructure; import org.apache.ignite.internal.pagememory.io.PageIo; import org.apache.ignite.internal.pagememory.reuse.LongListReuseBag; @@ -50,7 +50,7 @@ public class BlobStorage extends DataStructure { /** * Creates a new instance. */ - public BlobStorage(ReuseList reuseList, PageMemory pageMemory, int groupId, int partitionId) { + public BlobStorage(ReuseList reuseList, PartitionPageMemory pageMemory, int groupId, int partitionId) { super("BlobStorage", groupId, null, partitionId, pageMemory, PageIdAllocator.FLAG_AUX); super.reuseList = reuseList; diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/IndexStorageFactory.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/IndexStorageFactory.java index ecb4c925c14f..7d84ab2bc4bf 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/IndexStorageFactory.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/IndexStorageFactory.java @@ -22,7 +22,7 @@ import java.util.UUID; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.pagememory.PageIdAllocator; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.freelist.FreeListImpl; import org.apache.ignite.internal.storage.StorageException; import org.apache.ignite.internal.storage.index.StorageHashIndexDescriptor; @@ -45,6 +45,8 @@ class IndexStorageFactory { private final int partitionId; + private final PartitionPageMemory partitionPageMemory; + private final IndexMetaTree indexMetaTree; private final FreeListImpl freeList; @@ -67,12 +69,13 @@ private static class IndexTreeAndMeta { IndexStorageFactory( AbstractPageMemoryTableStorage tableStorage, - int partitionId, + PartitionPageMemory partitionPageMemory, IndexMetaTree indexMetaTree, FreeListImpl freeList ) { this.tableStorage = tableStorage; - this.partitionId = partitionId; + this.partitionId = partitionPageMemory.partitionId(); + this.partitionPageMemory = partitionPageMemory; this.indexMetaTree = indexMetaTree; this.freeList = freeList; } @@ -128,7 +131,7 @@ private IndexTreeAndMeta createHashIndexTreeAndMeta(StorageHashIn tableStorage.getTableId(), Integer.toString(tableStorage.getTableId()), partitionId, - tableStorage.dataRegion().pageMemory(), + partitionPageMemory, tableStorage.engine().generateGlobalRemoveId(), metaPageId, freeList, @@ -142,7 +145,7 @@ private HashIndexTree restoreHashIndexTree(IndexMeta indexMeta) { tableStorage.getTableId(), Integer.toString(tableStorage.getTableId()), partitionId, - tableStorage.dataRegion().pageMemory(), + partitionPageMemory, tableStorage.engine().generateGlobalRemoveId(), indexMeta.metaPageId(), freeList @@ -210,7 +213,7 @@ private IndexTreeAndMeta createSortedIndexTreeAndMeta(StorageSo tableStorage.getTableId(), Integer.toString(tableStorage.getTableId()), partitionId, - tableStorage.dataRegion().pageMemory(), + partitionPageMemory, tableStorage.engine().generateGlobalRemoveId(), metaPageId, freeList, @@ -226,7 +229,7 @@ private SortedIndexTree restoreSortedIndexTree(StorageSortedIndexDescriptor inde tableStorage.getTableId(), Integer.toString(tableStorage.getTableId()), partitionId, - tableStorage.dataRegion().pageMemory(), + partitionPageMemory, tableStorage.engine().generateGlobalRemoveId(), indexMeta.metaPageId(), freeList, @@ -244,7 +247,7 @@ private SortedIndexTree restoreSortedIndexTreeForDestroy(IndexMeta indexMeta) { tableStorage.getTableId(), Integer.toString(tableStorage.getTableId()), partitionId, - tableStorage.dataRegion().pageMemory(), + partitionPageMemory, tableStorage.engine().generateGlobalRemoveId(), indexMeta.metaPageId(), freeList @@ -274,9 +277,7 @@ void updateDataStructuresIn(PageMemorySortedIndexStorage indexStorage) { private IndexTreeAndMeta createIndexTree(StorageIndexDescriptor descriptor, IndexTreeConstructor treeConstructor) { try { - PageMemory pageMemory = tableStorage.dataRegion().pageMemory(); - - long metaPageId = pageMemory.allocatePage(freeList, tableStorage.getTableId(), partitionId, PageIdAllocator.FLAG_AUX); + long metaPageId = partitionPageMemory.allocatePage(freeList, tableStorage.getTableId(), partitionId, PageIdAllocator.FLAG_AUX); T tree = treeConstructor.createTree(metaPageId); diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PersistentPageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PersistentPageMemoryMvPartitionStorage.java index 8619e537d4b8..d6626e77baf0 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PersistentPageMemoryMvPartitionStorage.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PersistentPageMemoryMvPartitionStorage.java @@ -37,6 +37,7 @@ import org.apache.ignite.internal.hlc.HybridTimestamp; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.pagememory.DataRegion; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.freelist.FreeListImpl; import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory; import org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointListener; @@ -119,6 +120,7 @@ public class PersistentPageMemoryMvPartitionStorage extends AbstractPageMemoryMv public PersistentPageMemoryMvPartitionStorage( PersistentPageMemoryTableStorage tableStorage, int partitionId, + PartitionPageMemory partitionPageMemory, StoragePartitionMeta meta, FreeListImpl freeList, VersionChainTree versionChainTree, @@ -133,7 +135,7 @@ public PersistentPageMemoryMvPartitionStorage( tableStorage, new RenewablePartitionStorageState( tableStorage, - partitionId, + partitionPageMemory, versionChainTree, freeList, indexMetaTree, @@ -166,7 +168,7 @@ public void onMarkCheckpointBegin(CheckpointProgress progress, @Nullable Executo blobStorage = new BlobStorage( freeList, - dataRegion.pageMemory(), + partitionPageMemory, tableStorage.getTableId(), partitionId ); @@ -590,6 +592,7 @@ public void lastAppliedOnRebalance(long lastAppliedIndex, long lastAppliedTerm) * @throws StorageException If failed. */ public void updateDataStructures( + PartitionPageMemory partitionPageMemory, StoragePartitionMeta meta, FreeListImpl freeList, VersionChainTree versionChainTree, @@ -602,19 +605,20 @@ public void updateDataStructures( this.blobStorage = new BlobStorage( freeList, - tableStorage.dataRegion().pageMemory(), + partitionPageMemory, tableStorage.getTableId(), partitionId ); updateRenewableState( + partitionPageMemory, versionChainTree, freeList, indexMetaTree, gcQueue ); - checkpointManager.addCheckpointListener(checkpointListener, tableStorage.dataRegion()); + checkpointManager.addCheckpointListener(checkpointListener, (DataRegion) tableStorage.dataRegion()); } @Override @@ -704,7 +708,7 @@ WriteIntentLinks readWriteIntentLinks(long rowVersionLink) { var read = new ReadWriteIntentLinks(partitionId); try { - rowVersionDataPageReader.traverse(rowVersionLink, read, null); + renewableState.rowVersionDataPageReader().traverse(rowVersionLink, read, null); } catch (IgniteInternalCheckedException e) { throw new StorageException("Write intent links lookup failed: [link={}, {}]", e, rowVersionLink, createStorageInfo()); } diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/RenewablePartitionStorageState.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/RenewablePartitionStorageState.java index ccff1b33701e..9043fdee6431 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/RenewablePartitionStorageState.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/RenewablePartitionStorageState.java @@ -17,6 +17,8 @@ package org.apache.ignite.internal.storage.pagememory.mv; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; +import org.apache.ignite.internal.pagememory.datapage.DataPageReader; import org.apache.ignite.internal.pagememory.freelist.FreeListImpl; import org.apache.ignite.internal.storage.pagememory.AbstractPageMemoryTableStorage; import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaTree; @@ -37,10 +39,12 @@ class RenewablePartitionStorageState { private final IndexStorageFactory indexStorageFactory; + private final DataPageReader rowVersionDataPageReader; + /** Creates a new instance. */ RenewablePartitionStorageState( - AbstractPageMemoryTableStorage tableStorage, - int partitionId, + AbstractPageMemoryTableStorage tableStorage, + PartitionPageMemory partitionPageMemory, VersionChainTree versionChainTree, FreeListImpl freeList, IndexMetaTree indexMetaTree, @@ -53,10 +57,12 @@ class RenewablePartitionStorageState { this.indexStorageFactory = new IndexStorageFactory( tableStorage, - partitionId, + partitionPageMemory, indexMetaTree, freeList ); + + this.rowVersionDataPageReader = new DataPageReader(partitionPageMemory, tableStorage.getTableId()); } VersionChainTree versionChainTree() { @@ -78,4 +84,8 @@ GcQueue gcQueue() { IndexStorageFactory indexStorageFactory() { return indexStorageFactory; } + + public DataPageReader rowVersionDataPageReader() { + return rowVersionDataPageReader; + } } diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChainTree.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChainTree.java index 8b7f14e411dd..0ef2b2c34640 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChainTree.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VersionChainTree.java @@ -19,7 +19,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.reuse.ReuseList; import org.apache.ignite.internal.pagememory.tree.BplusTree; import org.apache.ignite.internal.pagememory.tree.io.BplusIo; @@ -52,7 +52,7 @@ public VersionChainTree( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList, diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VolatilePageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VolatilePageMemoryMvPartitionStorage.java index 8997e2c1bb96..6d31b917f842 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VolatilePageMemoryMvPartitionStorage.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VolatilePageMemoryMvPartitionStorage.java @@ -37,6 +37,7 @@ import org.apache.ignite.internal.hlc.HybridTimestamp; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.lang.IgniteInternalException; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.tree.BplusTree; import org.apache.ignite.internal.pagememory.util.GradualTask; import org.apache.ignite.internal.pagememory.util.GradualTaskExecutor; @@ -81,7 +82,7 @@ public class VolatilePageMemoryMvPartitionStorage extends AbstractPageMemoryMvPa * Constructor. * * @param tableStorage Table storage instance. - * @param partitionId Partition id. + * @param partitionPageMemory Page memory instance for a specific partition. * @param versionChainTree Table tree for {@link VersionChain}. * @param indexMetaTree Tree that contains SQL indexes' metadata. * @param destructionExecutor Executor used to destruct partitions. @@ -90,7 +91,7 @@ public class VolatilePageMemoryMvPartitionStorage extends AbstractPageMemoryMvPa */ public VolatilePageMemoryMvPartitionStorage( VolatilePageMemoryTableStorage tableStorage, - int partitionId, + PartitionPageMemory partitionPageMemory, VersionChainTree versionChainTree, IndexMetaTree indexMetaTree, GcQueue gcQueue, @@ -98,11 +99,11 @@ public VolatilePageMemoryMvPartitionStorage( FailureProcessor failureProcessor ) { super( - partitionId, + partitionPageMemory.partitionId(), tableStorage, new RenewablePartitionStorageState( tableStorage, - partitionId, + partitionPageMemory, versionChainTree, tableStorage.dataRegion().freeList(), indexMetaTree, @@ -347,6 +348,7 @@ List getResourcesToCloseOnCleanup() { * @throws StorageException If failed. */ public void updateDataStructures( + PartitionPageMemory partitionPageMemory, VersionChainTree versionChainTree, IndexMetaTree indexMetaTree, GcQueue gcQueue @@ -356,6 +358,7 @@ public void updateDataStructures( RenewablePartitionStorageState prevState = this.renewableState; updateRenewableState( + partitionPageMemory, versionChainTree, prevState.freeList(), indexMetaTree, diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/gc/GcQueue.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/gc/GcQueue.java index 8e50dbbb20df..0cbec97c0c27 100644 --- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/gc/GcQueue.java +++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/gc/GcQueue.java @@ -20,7 +20,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.ignite.internal.hlc.HybridTimestamp; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; -import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.reuse.ReuseList; import org.apache.ignite.internal.pagememory.tree.BplusTree; import org.apache.ignite.internal.pagememory.tree.io.BplusIo; @@ -55,7 +55,7 @@ public GcQueue( int grpId, String grpName, int partId, - PageMemory pageMem, + PartitionPageMemory pageMem, AtomicLong globalRmvId, long metaPageId, @Nullable ReuseList reuseList, diff --git a/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryNoLoadTest.java b/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryNoLoadTest.java index c11a1b0f2fda..49a54f5b176e 100644 --- a/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryNoLoadTest.java +++ b/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryNoLoadTest.java @@ -66,6 +66,7 @@ import org.apache.ignite.internal.pagememory.FullPageId; import org.apache.ignite.internal.pagememory.PageIdAllocator; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.TestDataRegion; import org.apache.ignite.internal.pagememory.configuration.CheckpointConfiguration; import org.apache.ignite.internal.pagememory.configuration.PersistentDataRegionConfiguration; @@ -172,6 +173,8 @@ void testDirtyPages(@WorkDirectory Path workDir) throws Exception { shouldNotHappenFlushDirtyPageForReplacement() ); + PartitionPageMemory partitionPageMemory = pageMemory.createPartitionPageMemory(GRP_ID, PARTITION_ID); + dataRegions.add(new TestDataRegion<>(pageMemory)); filePageStoreManager.start(); @@ -184,12 +187,15 @@ void testDirtyPages(@WorkDirectory Path workDir) throws Exception { checkpointManager.checkpointTimeoutLock().checkpointReadLock(); try { - Set dirtyPages = Set.of(createDirtyPage(pageMemory), createDirtyPage(pageMemory)); + Set dirtyPages = Set.of(createDirtyPage(partitionPageMemory), createDirtyPage(partitionPageMemory)); assertThat(pageMemory.dirtyPages(), equalTo(dirtyPages)); assertEquals(2, pageMemory.invalidate(GRP_ID, PARTITION_ID)); - Set dirtyPagesAfterInvalidation = Set.of(createDirtyPage(pageMemory), createDirtyPage(pageMemory)); + Set dirtyPagesAfterInvalidation = Set.of( + createDirtyPage(partitionPageMemory), + createDirtyPage(partitionPageMemory) + ); assertThat(pageMemory.dirtyPages(), equalTo(union(dirtyPages, dirtyPagesAfterInvalidation))); } finally { checkpointManager.checkpointTimeoutLock().checkpointReadUnlock(); @@ -237,6 +243,8 @@ void testCheckpointUrgency(@WorkDirectory Path workDir) throws Exception { shouldNotHappenFlushDirtyPageForReplacement() ); + PartitionPageMemory partitionPageMemory = pageMemory.createPartitionPageMemory(GRP_ID, PARTITION_ID); + dataRegions.add(new TestDataRegion<>(pageMemory)); filePageStoreManager.start(); @@ -260,19 +268,19 @@ void testCheckpointUrgency(@WorkDirectory Path workDir) throws Exception { int i = 0; for (; i < dirtyPagesSoftThreshold - 1; i++) { - createDirtyPage(pageMemory); + createDirtyPage(partitionPageMemory); assertEquals(NOT_REQUIRED, pageMemory.checkpointUrgency(), "i=" + i); } for (; i < dirtyPagesHardThreshold - 1; i++) { - createDirtyPage(pageMemory); + createDirtyPage(partitionPageMemory); assertEquals(SHOULD_TRIGGER, pageMemory.checkpointUrgency(), "i=" + i); } for (; i < maxPages; i++) { - createDirtyPage(pageMemory); + createDirtyPage(partitionPageMemory); assertEquals(MUST_TRIGGER, pageMemory.checkpointUrgency(), "i=" + i); } @@ -318,6 +326,8 @@ void testDeltaFilePageStore(@WorkDirectory Path workDir) throws Exception { shouldNotHappenFlushDirtyPageForReplacement() ); + PartitionPageMemory partitionPageMemory = pageMemory.createPartitionPageMemory(GRP_ID, PARTITION_ID); + dataRegions.add(new TestDataRegion<>(pageMemory)); filePageStoreManager.start(); @@ -330,9 +340,9 @@ void testDeltaFilePageStore(@WorkDirectory Path workDir) throws Exception { checkpointManager.checkpointTimeoutLock().checkpointReadLock(); try { - createDirtyPage(pageMemory); - createDirtyPage(pageMemory); - createDirtyPage(pageMemory); + createDirtyPage(partitionPageMemory); + createDirtyPage(partitionPageMemory); + createDirtyPage(partitionPageMemory); } finally { checkpointManager.checkpointTimeoutLock().checkpointReadUnlock(); } @@ -383,6 +393,8 @@ void testPageReplacement(@WorkDirectory Path workDir) throws Exception { } ); + PartitionPageMemory partitionPageMemory = pageMemory.createPartitionPageMemory(GRP_ID, PARTITION_ID); + dataRegions.add(new TestDataRegion<>(pageMemory)); filePageStoreManager.start(); @@ -410,7 +422,7 @@ void testPageReplacement(@WorkDirectory Path workDir) throws Exception { try { for (int i = 0; i < 1_000; i++) { - createDirtyPage(pageMemory); + createDirtyPage(partitionPageMemory); } } finally { checkpointManager.checkpointTimeoutLock().checkpointReadUnlock(); @@ -429,7 +441,7 @@ void testPageReplacement(@WorkDirectory Path workDir) throws Exception { do { // We create new dirty pages so that we get to the end of the data region and start page replacing. - createDirtyPage(pageMemory); + createDirtyPage(partitionPageMemory); } while (!flushDirtyPageForReplacementFuture.isDone()); // Let's write the dirty pages to disk and complete the checkpoint. @@ -451,7 +463,7 @@ void testPageReplacement(@WorkDirectory Path workDir) throws Exception { } /** - * Tests that {@link PersistentPageMemory#acquirePage(int, long)} works correctly when multiple threads try to acquire the same page + * Tests that {@link PartitionPageMemory#acquirePage(int, long)} works correctly when multiple threads try to acquire the same page * using different {@code pageId} values, assuming that one of the threads simply has an invalid identifier from some obsolete source. */ @Test @@ -482,6 +494,8 @@ void testAcquireRace(@WorkDirectory Path workDir) throws Exception { shouldNotHappenFlushDirtyPageForReplacement() ); + PartitionPageMemory partitionPageMemory = pageMemory.createPartitionPageMemory(GRP_ID, PARTITION_ID); + dataRegions.add(new TestDataRegion<>(pageMemory)); filePageStoreManager.start(); @@ -494,7 +508,7 @@ void testAcquireRace(@WorkDirectory Path workDir) throws Exception { try { for (int i = 0; i < pages; i++) { - createDirtyPage(pageMemory); + createDirtyPage(partitionPageMemory); } } finally { checkpointManager.checkpointTimeoutLock().checkpointReadUnlock(); @@ -522,6 +536,7 @@ void testAcquireRace(@WorkDirectory Path workDir) throws Exception { checkpointManager, shouldNotHappenFlushDirtyPageForReplacement() ); + PartitionPageMemory partitionPageMemory2 = pageMemory2.createPartitionPageMemory(GRP_ID, PARTITION_ID); filePageStoreManager.start(); checkpointManager.start(); @@ -537,11 +552,11 @@ void testAcquireRace(@WorkDirectory Path workDir) throws Exception { // Step 3. Run the race for all pages in the partition. // It's fine to not release/unlock these pages, we stop the region immediately after. IgniteTestUtils.runRace( - () -> pageMemory2.acquirePage(GRP_ID, fakePageId), + () -> partitionPageMemory2.acquirePage(GRP_ID, fakePageId), () -> { - long page = pageMemory2.acquirePage(GRP_ID, realPageId); + long page = partitionPageMemory2.acquirePage(GRP_ID, realPageId); - assertNotEquals(0L, pageMemory2.readLock(GRP_ID, realPageId, page)); + assertNotEquals(0L, partitionPageMemory2.readLock(GRP_ID, realPageId, page)); } ); } @@ -575,7 +590,7 @@ private PersistentPageMemory createPageMemory( ); } - private DirtyFullPageId createDirtyPage(PersistentPageMemory pageMemory) throws Exception { + private DirtyFullPageId createDirtyPage(PartitionPageMemory pageMemory) throws Exception { FullPageId fullPageId = allocatePage(pageMemory); long page = pageMemory.acquirePage(fullPageId.groupId(), fullPageId.pageId()); @@ -691,12 +706,14 @@ private static void initGroupFilePageStores( public void testLoadedPagesCount() { PageMemory mem = memory(); + PartitionPageMemory partitionPageMemory = mem.createPartitionPageMemory(GRP_ID, PARTITION_ID); + int expPages = MAX_MEMORY_SIZE / mem.systemPageSize(); try { assertDoesNotThrow(() -> { for (int i = 0; i < expPages * 2; i++) { - allocatePage(mem); + allocatePage(partitionPageMemory); } }); } finally { @@ -707,10 +724,12 @@ public void testLoadedPagesCount() { @Test void testPartitionGenerationAfterAllocatePage() throws Exception { runWithStartedPersistentPageMemory(mem -> { - FullPageId fullPageId = allocatePage(mem); + PartitionPageMemory partitionPageMemory = mem.createPartitionPageMemory(GRP_ID, PARTITION_ID); + + FullPageId fullPageId = allocatePage(partitionPageMemory); // Absolute memory pointer to page with header. - long absPtr = mem.acquirePage(fullPageId.groupId(), fullPageId.pageId()); + long absPtr = partitionPageMemory.acquirePage(fullPageId.groupId(), fullPageId.pageId()); assertEquals(1, partitionGeneration(absPtr)); }); @@ -721,10 +740,12 @@ void testPartitionGenerationAfterAllocatePageAndInvalidatePartition() throws Exc runWithStartedPersistentPageMemory(mem -> { assertEquals(2, mem.invalidate(GRP_ID, PARTITION_ID)); - FullPageId fullPageId = allocatePage(mem); + PartitionPageMemory partitionPageMemory = mem.createPartitionPageMemory(GRP_ID, PARTITION_ID); + + FullPageId fullPageId = allocatePage(partitionPageMemory); // Absolute memory pointer to page with header. - long absPtr = mem.acquirePage(fullPageId.groupId(), fullPageId.pageId()); + long absPtr = partitionPageMemory.acquirePage(fullPageId.groupId(), fullPageId.pageId()); assertEquals(2, partitionGeneration(absPtr)); }); @@ -733,7 +754,9 @@ void testPartitionGenerationAfterAllocatePageAndInvalidatePartition() throws Exc @Test void testPartitionGenerationAfterCheckpointWritePageAndInvalidatePartition() throws Exception { runWithStartedPersistentPageMemory(mem -> { - DirtyFullPageId fullPageId = allocateDirtyPage(mem); + PartitionPageMemory partitionPageMemory = mem.createPartitionPageMemory(GRP_ID, PARTITION_ID); + + DirtyFullPageId fullPageId = allocateDirtyPage(partitionPageMemory); mem.beginCheckpoint(new CheckpointProgressImpl(42)); @@ -747,8 +770,10 @@ void testPartitionGenerationAfterCheckpointWritePageAndInvalidatePartition() thr true ); + partitionPageMemory = mem.createPartitionPageMemory(GRP_ID, PARTITION_ID); + // Absolute memory pointer to page with header. - long absPtr = mem.acquirePage(fullPageId.groupId(), fullPageId.pageId()); + long absPtr = partitionPageMemory.acquirePage(fullPageId.groupId(), fullPageId.pageId()); assertEquals(2, partitionGeneration(absPtr)); }); @@ -775,7 +800,7 @@ private interface ConsumerX { * @param mem Memory. * @throws IgniteInternalCheckedException If failed. */ - public static DirtyFullPageId allocateDirtyPage(PersistentPageMemory mem) throws IgniteInternalCheckedException { + public static DirtyFullPageId allocateDirtyPage(PartitionPageMemory mem) throws IgniteInternalCheckedException { long pageId = mem.allocatePageNoReuse(GRP_ID, PARTITION_ID, PageIdAllocator.FLAG_DATA); long page = mem.acquirePage(GRP_ID, pageId); diff --git a/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/benchmarks/SortedIndexTreeInsertBenchmark.java b/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/benchmarks/SortedIndexTreeInsertBenchmark.java index c9ddadba64ef..468dfcefc02d 100644 --- a/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/benchmarks/SortedIndexTreeInsertBenchmark.java +++ b/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/benchmarks/SortedIndexTreeInsertBenchmark.java @@ -26,6 +26,7 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.function.Supplier; import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.benchmark.VolatilePageMemoryBenchmarkBase; import org.apache.ignite.internal.storage.RowId; import org.apache.ignite.internal.storage.index.StorageSortedIndexDescriptor; @@ -118,13 +119,14 @@ public void setup() throws Exception { StorageSortedIndexDescriptor indexDescriptor = new StorageSortedIndexDescriptor(INDEX_ID, columnTypes.columnDescriptors(), false); + PartitionPageMemory partitionPageMemory = volatilePageMemory.createPartitionPageMemory(GROUP_ID, PARTITION_ID); sortedIndexTree = SortedIndexTree.createNew( GROUP_ID, "sortedIndex", PARTITION_ID, - volatilePageMemory, + partitionPageMemory, new AtomicLong(), - volatilePageMemory.allocatePageNoReuse(GROUP_ID, PARTITION_ID, FLAG_AUX), + partitionPageMemory.allocatePageNoReuse(GROUP_ID, PARTITION_ID, FLAG_AUX), freeList, indexDescriptor, createNewJitComparator(indexDescriptor) diff --git a/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/mv/BlobStorageTest.java b/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/mv/BlobStorageTest.java index 6b87edf47d0b..c166ec8ca66b 100644 --- a/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/mv/BlobStorageTest.java +++ b/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/mv/BlobStorageTest.java @@ -33,6 +33,7 @@ import java.util.List; import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension; import org.apache.ignite.internal.pagememory.PageMemory; +import org.apache.ignite.internal.pagememory.PartitionPageMemory; import org.apache.ignite.internal.pagememory.configuration.VolatileDataRegionConfiguration; import org.apache.ignite.internal.pagememory.inmemory.VolatilePageMemory; import org.apache.ignite.internal.pagememory.io.PageIoRegistry; @@ -41,6 +42,7 @@ import org.apache.ignite.internal.storage.pagememory.mv.io.BlobFragmentIo; import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest; import org.apache.ignite.internal.util.OffheapReadWriteLock; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -59,6 +61,8 @@ class BlobStorageTest extends BaseIgniteAbstractTest { private PageMemory pageMemory; + private PartitionPageMemory partitionPageMemory; + @Captor private ArgumentCaptor reuseBagCaptor; @@ -78,7 +82,14 @@ void createStorage() { new OffheapReadWriteLock(OffheapReadWriteLock.DEFAULT_CONCURRENCY_LEVEL) )); - blobStorage = new BlobStorage(reuseList, pageMemory, 1, 1); + partitionPageMemory = spy(pageMemory.createPartitionPageMemory(1, 1)); + + blobStorage = new BlobStorage(reuseList, partitionPageMemory, 1, 1); + } + + @AfterEach + void destroyStorage() { + pageMemory.stop(true); } @Test @@ -92,7 +103,7 @@ void attemptsReuseOnAddition() throws Exception { void allocatesOnlyRequiredNumberOfPages() throws Exception { blobStorage.addBlob(new byte[(int) (PAGE_SIZE * 1.5)]); - verify(pageMemory, times(2)).allocatePageNoReuse(anyInt(), anyInt(), anyByte()); + verify(partitionPageMemory, times(2)).allocatePageNoReuse(anyInt(), anyInt(), anyByte()); } @Test @@ -101,7 +112,7 @@ void allocatesOnlyRequiredNumberOfPagesOnUpdate() throws Exception { blobStorage.updateBlob(pageId, new byte[(int) (PAGE_SIZE * 2.5)]); - verify(pageMemory, times(3)).allocatePageNoReuse(anyInt(), anyInt(), anyByte()); + verify(partitionPageMemory, times(3)).allocatePageNoReuse(anyInt(), anyInt(), anyByte()); } @Test @@ -110,7 +121,7 @@ void doesNotAllocateOnShortening() throws Exception { blobStorage.updateBlob(pageId, new byte[(int) (PAGE_SIZE * 0.5)]); - verify(pageMemory, times(2)).allocatePageNoReuse(anyInt(), anyInt(), anyByte()); + verify(partitionPageMemory, times(2)).allocatePageNoReuse(anyInt(), anyInt(), anyByte()); } @Test @@ -156,7 +167,7 @@ void freedPagesAreRecycled() throws Exception { blobStorage.updateBlob(pageId, new byte[0]); - verify(pageMemory, times(3)).allocatePageNoReuse(anyInt(), anyInt(), anyByte()); + verify(partitionPageMemory, times(3)).allocatePageNoReuse(anyInt(), anyInt(), anyByte()); verify(reuseList).addForRecycle(reuseBagCaptor.capture());