IGNITE-26252 wip
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 2f1a497..0d271a6 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
@@ -33,6 +33,7 @@ import static org.apache.ignite.internal.pagememory.persistence.PageHeader.fullPageId; import static org.apache.ignite.internal.pagememory.persistence.PageHeader.isAcquired; import static org.apache.ignite.internal.pagememory.persistence.PageHeader.partitionGeneration; +import static org.apache.ignite.internal.pagememory.persistence.PageHeader.pinCount; import static org.apache.ignite.internal.pagememory.persistence.PageHeader.tempBufferPointer; import static org.apache.ignite.internal.pagememory.persistence.PageHeader.timestamp; import static org.apache.ignite.internal.pagememory.persistence.PagePool.SEGMENT_INDEX_MASK; @@ -60,13 +61,13 @@ import static org.apache.ignite.internal.util.StringUtils.hexLong; import static org.apache.ignite.internal.util.StringUtils.toHexString; +import it.unimi.dsi.fastutil.objects.Object2IntMap; +import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collection; -import java.util.HashMap; import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadLocalRandom; @@ -530,10 +531,10 @@ @Override public long allocatePageNoReuse(int grpId, int partId, byte flags) throws IgniteInternalCheckedException { - assert partId >= 0 && partId <= MAX_PARTITION_ID : partId; + assert partId >= 0 && partId <= MAX_PARTITION_ID : "grpId=" + grpId + ", partId=" + partId; - assert started; - assert checkpointTimeoutLock.checkpointLockIsHeldByThread(); + assert started : "grpId=" + grpId + ", partId=" + partId; + assert checkpointTimeoutLock.checkpointLockIsHeldByThread() : "grpId=" + grpId + ", partId=" + partId; PagesWriteThrottlePolicy writeThrottle = this.writeThrottle; if (writeThrottle != null) { @@ -552,10 +553,12 @@ try { FullPageId fullId = new FullPageId(pageId, grpId); + int partGen = partGeneration(seg, fullId); + long relPtr = seg.loadedPages.get( grpId, effectivePageId(pageId), - seg.partGeneration(grpId, partId), + partGen, INVALID_REL_PTR, OUTDATED_REL_PTR ); @@ -580,21 +583,22 @@ fullPageId(absPtr, fullId); timestamp(absPtr, coarseCurrentTimeMillis()); - partitionGeneration(absPtr, seg.partGeneration(grpId, partId)); + partitionGeneration(absPtr, partGen); rwLock.init(absPtr + PAGE_LOCK_OFFSET, tag(pageId)); - assert getCrc(absPtr + PAGE_OVERHEAD) == 0; // TODO IGNITE-16612 + assert getCrc(absPtr + PAGE_OVERHEAD) == 0 : fullId; // TODO IGNITE-16612 - assert !isAcquired(absPtr) : - "Pin counter must be 0 for a new page [relPtr=" + hexLong(relPtr) - + ", absPtr=" + hexLong(absPtr) + ", pinCntr=" + PageHeader.pinCount(absPtr) + ']'; + assert !isAcquired(absPtr) : String.format( + "Pin counter must be 0 for a new page [relPtr=%s, absPtr=%s, pinCntr=%s, fullId=%s]", + hexLong(relPtr), hexLong(absPtr), pinCount(absPtr), fullId + ); setDirty(fullId, absPtr, true, true); seg.pageReplacementPolicy.onMiss(relPtr); - seg.loadedPages.put(grpId, effectivePageId(pageId), relPtr, seg.partGeneration(grpId, partId)); + seg.loadedPages.put(grpId, effectivePageId(pageId), relPtr, partGen); } catch (IgniteOutOfMemoryException oom) { IgniteOutOfMemoryException e = new IgniteOutOfMemoryException("Out of memory in data region [" + "name=" + dataRegionConfiguration.name() @@ -671,10 +675,13 @@ boolean restore, @Nullable AtomicBoolean pageAllocated ) throws IgniteInternalCheckedException { - assert started; - assert pageIndex(pageId) != 0 : "Partition meta should should not be read through PageMemory so as not to occupy memory."; + 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]", + grpId, hexLong(pageId) + ); - int partId = partitionId(pageId); + FullPageId fullId = new FullPageId(pageId, grpId); Segment seg = segment(grpId, pageId); @@ -684,7 +691,7 @@ long relPtr = seg.loadedPages.get( grpId, effectivePageId(pageId), - seg.partGeneration(grpId, partId), + partGeneration(seg, fullId), INVALID_REL_PTR, INVALID_REL_PTR ); @@ -703,19 +710,19 @@ seg.readLock().unlock(); } - FullPageId fullId = new FullPageId(pageId, grpId); - seg.writeLock().lock(); long lockedPageAbsPtr = -1; boolean readPageFromStore = false; try { + int partGen = partGeneration(seg, fullId); + // Double-check. long relPtr = seg.loadedPages.get( grpId, fullId.effectivePageId(), - seg.partGeneration(grpId, partId), + partGen, INVALID_REL_PTR, OUTDATED_REL_PTR ); @@ -737,10 +744,12 @@ fullPageId(absPtr, fullId); timestamp(absPtr, coarseCurrentTimeMillis()); - partitionGeneration(absPtr, seg.partGeneration(grpId, partId)); + partitionGeneration(absPtr, partGen); - assert !isAcquired(absPtr) : - "Pin counter must be 0 for a new page [relPtr=" + hexLong(relPtr) + ", absPtr=" + hexLong(absPtr) + ']'; + assert !isAcquired(absPtr) : String.format( + "Pin counter must be 0 for a new page [relPtr=%s, absPtr=%s, pinCntr=%s, fullId=%s]", + hexLong(relPtr), hexLong(absPtr), pinCount(absPtr), fullId + ); // We can clear dirty flag after the page has been allocated. setDirty(fullId, absPtr, false, false); @@ -751,7 +760,7 @@ grpId, fullId.effectivePageId(), relPtr, - seg.partGeneration(grpId, partId) + partGen ); long pageAddr = absPtr + PAGE_OVERHEAD; @@ -789,11 +798,13 @@ fullPageId(absPtr, fullId); timestamp(absPtr, coarseCurrentTimeMillis()); - partitionGeneration(absPtr, seg.partGeneration(grpId, partId)); + partitionGeneration(absPtr, partGen); setPageId(pageAddr, pageId); - assert !isAcquired(absPtr) : - "Pin counter must be 0 for a new page [relPtr=" + hexLong(relPtr) + ", absPtr=" + hexLong(absPtr) + ']'; + assert !isAcquired(absPtr) : String.format( + "Pin counter must be 0 for a new page [relPtr=%s, absPtr=%s, pinCntr=%s, fullId=%s]", + hexLong(relPtr), hexLong(absPtr), pinCount(absPtr), fullId + ); rwLock.init(absPtr + PAGE_LOCK_OFFSET, tag(pageId)); @@ -890,17 +901,13 @@ } /** - * Get current partition generation tag. + * Get current partition generation. * * @param seg Segment. - * @param fullPageId Full page id. - * @return Current partition generation tag. + * @param fullPageId Full page ID. */ - private int generationTag(Segment seg, FullPageId fullPageId) { - return seg.partGeneration( - fullPageId.groupId(), - partitionId(fullPageId.pageId()) - ); + private int partGeneration(Segment seg, FullPageId fullPageId) { + return seg.partGeneration(fullPageId.groupId(), fullPageId.partitionId()); } /** @@ -929,7 +936,8 @@ * * @param grpId Group ID. * @param partId Partition ID. - * @return New partition generation (growing 1-based partition file version). + * @return New partition generation (growing 1-based partition file version), {@code 0} if the {@link PersistentPageMemory} did not + * start or stopped. */ public int invalidate(int grpId, int partId) { synchronized (segmentsLock) { @@ -937,25 +945,28 @@ return 0; } - int tag = 0; + int resultPartitionGeneration = 0; for (Segment segment : segments) { segment.writeLock().lock(); try { - int newTag = segment.incrementPartGeneration(grpId, partId); + int newPartitionGeneration = segment.incrementPartGeneration(grpId, partId); - if (tag == 0) { - tag = newTag; + if (resultPartitionGeneration == 0) { + resultPartitionGeneration = newPartitionGeneration; } - assert tag == newTag; + assert resultPartitionGeneration == newPartitionGeneration : String.format( + "grpId=%s, partId=%s, resultPartitionGeneration=%s, newPartitionGeneration=%s", + grpId, partId, resultPartitionGeneration, newPartitionGeneration + ); } finally { segment.writeLock().unlock(); } } - return tag; + return resultPartitionGeneration; } } @@ -1042,14 +1053,13 @@ public boolean hasLoadedPage(FullPageId fullPageId) { int grpId = fullPageId.groupId(); long pageId = fullPageId.effectivePageId(); - int partId = partitionId(pageId); Segment seg = segment(grpId, pageId); seg.readLock().lock(); try { - long res = seg.loadedPages.get(grpId, pageId, seg.partGeneration(grpId, partId), INVALID_REL_PTR, INVALID_REL_PTR); + long res = seg.loadedPages.get(grpId, pageId, partGeneration(seg, fullPageId), INVALID_REL_PTR, INVALID_REL_PTR); return res != INVALID_REL_PTR; } finally { @@ -1406,8 +1416,12 @@ /** Initial partition generation. */ private static final int INIT_PART_GENERATION = 1; - /** Maps partition (grpId, partId) to its generation. Generation is 1-based incrementing partition counter. */ - private final Map<GroupPartitionId, Integer> partGenerationMap = new HashMap<>(); + /** + * Maps partition (grpId, partId) to its generation. Generation is 1-based incrementing partition counter. + * + * <p>Guarded by segment read/write lock.</p> + */ + private final Object2IntMap<GroupPartitionId> partGenerationMap = new Object2IntOpenHashMap<>(); /** Segment closed flag. */ private boolean closed; @@ -1608,11 +1622,11 @@ * @return Relative pointer to refreshed page. */ public long refreshOutdatedPage(int grpId, long pageId, boolean rmv) { - assert writeLock().isHeldByCurrentThread(); + assert writeLock().isHeldByCurrentThread() : "grpId=" + grpId + ", pageId=" + pageId; - int tag = partGeneration(grpId, partitionId(pageId)); + int partGen = partGeneration(grpId, partitionId(pageId)); - long relPtr = loadedPages.refresh(grpId, effectivePageId(pageId), tag); + long relPtr = loadedPages.refresh(grpId, effectivePageId(pageId), partGen); long absPtr = absolute(relPtr); @@ -1722,13 +1736,15 @@ * @param partId Partition ID. */ public int partGeneration(int grpId, int partId) { - assert getReadHoldCount() > 0 || getWriteHoldCount() > 0; + assert getReadHoldCount() > 0 || getWriteHoldCount() > 0 : "grpId=" + grpId + ", partId=" + partId; - Integer tag = partGenerationMap.get(new GroupPartitionId(grpId, partId)); + var groupPartitionId = new GroupPartitionId(grpId, partId); - assert tag == null || tag >= 0 : "Negative tag=" + tag; + int partitionGeneration = partGenerationMap.getOrDefault(groupPartitionId, INIT_PART_GENERATION); - return tag == null ? INIT_PART_GENERATION : tag; + assert partitionGeneration > 0 : "groupPartitionId=" + groupPartitionId + ", partitionGeneration=" + partitionGeneration; + + return partitionGeneration; } /** @@ -1753,31 +1769,27 @@ * @return New partition generation. */ private int incrementPartGeneration(int grpId, int partId) { - assert getWriteHoldCount() > 0; + assert getWriteHoldCount() > 0 : "grpId=" + grpId + ", partId=" + partId; - GroupPartitionId grpPart = new GroupPartitionId(grpId, partId); + var groupPartitionId = new GroupPartitionId(grpId, partId); - Integer gen = partGenerationMap.get(grpPart); + int partitionGeneration = partGenerationMap.getOrDefault(groupPartitionId, INIT_PART_GENERATION); - if (gen == null) { - gen = INIT_PART_GENERATION; - } + if (partitionGeneration == Integer.MAX_VALUE) { + LOG.info("Partition generation overflow [grpId={}, partId={}]", grpId, partId); - if (gen == Integer.MAX_VALUE) { - LOG.info("Partition tag overflow [grpId={}, partId={}]", grpId, partId); + partGenerationMap.put(groupPartitionId, INIT_PART_GENERATION); - partGenerationMap.put(grpPart, 0); - - return 0; + return INIT_PART_GENERATION; } else { - partGenerationMap.put(grpPart, gen + 1); + partGenerationMap.put(groupPartitionId, partitionGeneration + 1); - return gen + 1; + return partitionGeneration + 1; } } private void resetGroupPartitionsGeneration(int grpId) { - assert getWriteHoldCount() > 0; + assert getWriteHoldCount() > 0 : "grpId=" + grpId; partGenerationMap.keySet().removeIf(grpPart -> grpPart.getGroupId() == grpId); } @@ -2035,7 +2047,7 @@ CheckpointMetricsTracker tracker, boolean useTryWriteLockOnPage ) throws IgniteInternalCheckedException { - assert buf.remaining() == pageSize() : "fullId=" + hexLong(fullId.pageId()) + ", remaining=" + buf.remaining(); + assert buf.remaining() == pageSize() : "fullId=" + fullId + ", remaining=" + buf.remaining(); Segment seg = segment(fullId.groupId(), fullId.pageId()); @@ -2043,7 +2055,7 @@ long relPtr; - int partitionGeneration; + int partGen; boolean pageSingleAcquire = false; @@ -2054,7 +2066,9 @@ return; } - relPtr = resolveRelativePointer(seg, fullId, partitionGeneration = generationTag(seg, fullId)); + partGen = partGeneration(seg, fullId); + + relPtr = resolveRelativePointer(seg, fullId, partGen); // Page may have been cleared during eviction. We have nothing to do in this case. if (relPtr == INVALID_REL_PTR) { @@ -2080,7 +2094,7 @@ try { // Double-check. - relPtr = resolveRelativePointer(seg, fullId, generationTag(seg, fullId)); + relPtr = resolveRelativePointer(seg, fullId, partGeneration(seg, fullId)); if (relPtr == INVALID_REL_PTR) { return; @@ -2108,7 +2122,7 @@ absPtr, fullId, buf, - partitionGeneration, + partGen, pageSingleAcquire, pageStoreWriter, tracker,
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/ClockPageReplacementPolicy.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/ClockPageReplacementPolicy.java index f14e80b..f2b643a 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/ClockPageReplacementPolicy.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/ClockPageReplacementPolicy.java
@@ -20,7 +20,6 @@ import static org.apache.ignite.internal.pagememory.persistence.PageHeader.fullPageId; import static org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory.INVALID_REL_PTR; import static org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory.OUTDATED_REL_PTR; -import static org.apache.ignite.internal.pagememory.util.PageIdUtils.partitionId; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.pagememory.FullPageId; @@ -80,12 +79,12 @@ relPtr = loadedPages.get( fullId.groupId(), fullId.effectivePageId(), - seg.partGeneration(fullId.groupId(), partitionId(fullId.pageId())), + seg.partGeneration(fullId.groupId(), fullId.partitionId()), INVALID_REL_PTR, OUTDATED_REL_PTR ); - assert relPtr != INVALID_REL_PTR; + assert relPtr != INVALID_REL_PTR : fullId; if (relPtr == OUTDATED_REL_PTR) { return seg.refreshOutdatedPage(fullId.groupId(), fullId.pageId(), true);
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/RandomLruPageReplacementPolicy.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/RandomLruPageReplacementPolicy.java index 774b608..542f06c 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/RandomLruPageReplacementPolicy.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/RandomLruPageReplacementPolicy.java
@@ -20,7 +20,6 @@ import static org.apache.ignite.internal.pagememory.persistence.PageHeader.PAGE_OVERHEAD; import static org.apache.ignite.internal.pagememory.persistence.PageHeader.fullPageId; import static org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory.INVALID_REL_PTR; -import static org.apache.ignite.internal.pagememory.util.PageIdUtils.partitionId; import java.util.HashSet; import java.util.Set; @@ -105,7 +104,7 @@ assert fullId.equals(nearest.fullId()) : "Invalid page mapping [tableId=" + nearest.fullId() + ", actual=" + fullId + ", nearest=" + nearest; - boolean outdated = partGen < seg.partGeneration(fullId.groupId(), partitionId(fullId.pageId())); + boolean outdated = partGen < seg.partGeneration(fullId.groupId(), fullId.partitionId()); if (outdated) { return seg.refreshOutdatedPage(fullId.groupId(), fullId.pageId(), true); @@ -222,7 +221,7 @@ FullPageId fullId = fullPageId(absPageAddr); - if (partGen < seg.partGeneration(fullId.groupId(), partitionId(fullId.pageId()))) { + if (partGen < seg.partGeneration(fullId.groupId(), fullId.partitionId())) { return seg.refreshOutdatedPage(fullId.groupId(), fullId.pageId(), true); }
diff --git a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/SegmentedLruPageReplacementPolicy.java b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/SegmentedLruPageReplacementPolicy.java index 541acd2..1120c3f 100644 --- a/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/SegmentedLruPageReplacementPolicy.java +++ b/modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/replacement/SegmentedLruPageReplacementPolicy.java
@@ -20,7 +20,6 @@ import static org.apache.ignite.internal.pagememory.persistence.PageHeader.fullPageId; import static org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory.INVALID_REL_PTR; import static org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory.OUTDATED_REL_PTR; -import static org.apache.ignite.internal.pagememory.util.PageIdUtils.partitionId; import org.apache.ignite.internal.lang.IgniteInternalCheckedException; import org.apache.ignite.internal.pagememory.FullPageId; @@ -88,12 +87,12 @@ relPtr = loadedPages.get( fullId.groupId(), fullId.effectivePageId(), - seg.partGeneration(fullId.groupId(), partitionId(fullId.pageId())), + seg.partGeneration(fullId.groupId(), fullId.partitionId()), INVALID_REL_PTR, OUTDATED_REL_PTR ); - assert relPtr != INVALID_REL_PTR; + assert relPtr != INVALID_REL_PTR : fullId; if (relPtr == OUTDATED_REL_PTR) { return seg.refreshOutdatedPage(fullId.groupId(), fullId.pageId(), true);