diff --git a/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java b/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java index ef77e535a..de2549ac0 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java @@ -49,7 +49,7 @@ public final class BigSegmentHeader implements SegmentHeader { static final int MAX_WAIT = Integer.MAX_VALUE; static final int WAIT_PARTY = 1; - static final long SIZE_OFFSET = LOCK_OFFSET + 8L; // 32-bit + static final long ENTRIES_OFFSET = LOCK_OFFSET + 8L; // 32-bit static final long LOWEST_POSSIBLY_FREE_CHUNK_OFFSET = ENTRIES_OFFSET + 4L; @@ -62,17 +62,17 @@ public final class BigSegmentHeader implements SegmentHeader { } @Override - public long size(long address) { - return OS.memory().readInt(address + SIZE_OFFSET) & UNSIGNED_INT_MASK; + public long entries(long address) { + return OS.memory().readInt(address + ENTRIES_OFFSET) & UNSIGNED_INT_MASK; } @Override - public void size(long address, long size) { - if (size >= (1L << 32)) { - throw new IllegalStateException("segment size overflow: up to " + UNSIGNED_INT_MASK + - " supported, " + size + " given"); + public void entries(long address, long entries) { + if (entries >= (1L << 32)) { + throw new IllegalStateException("segment entries overflow: up to " + UNSIGNED_INT_MASK + + " supported, " + entries + " given"); } - OS.memory().writeInt(address + SIZE_OFFSET, (int) size); + OS.memory().writeInt(address + ENTRIES_OFFSET, (int) entries); } @Override diff --git a/src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java b/src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java index d9876b725..32873becb 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java @@ -19,8 +19,8 @@ package net.openhft.chronicle.hash.impl; import java.util.concurrent.TimeUnit; public interface SegmentHeader { - long size(long address); - void size(long address, long size); + long entries(long address); + void entries(long address, long size); long deleted(long address); void deleted(long address, long deleted); diff --git a/src/main/java/net/openhft/chronicle/hash/impl/TierCountersArea.java b/src/main/java/net/openhft/chronicle/hash/impl/TierCountersArea.java index cee3666f0..da7614f63 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/TierCountersArea.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/TierCountersArea.java @@ -29,11 +29,15 @@ public enum TierCountersArea { ; private static Memory memory = OS.memory(); + private static final long UNSIGNED_INT_MASK = 0xFFFFFFFFL; + public static final long NEXT_TIER_INDEX_OFFSET = 0L; public static final long PREV_TIER_INDEX_OFFSET = NEXT_TIER_INDEX_OFFSET + 8L; public static final long LOWEST_POSSIBLY_FREE_CHUNK_TIERED_OFFSET = PREV_TIER_INDEX_OFFSET + 8L; public static final long SEGMENT_INDEX_OFFSET = LOWEST_POSSIBLY_FREE_CHUNK_TIERED_OFFSET + 8L; public static final long TIER_OFFSET = SEGMENT_INDEX_OFFSET + 4L; + public static final long ENTRIES_OFFSET = TIER_OFFSET + 4L; + public static final long DELETED_OFFSET = ENTRIES_OFFSET + 4L; public static long nextTierIndex(long address) { return memory.readLong(address + NEXT_TIER_INDEX_OFFSET); @@ -75,4 +79,28 @@ public enum TierCountersArea { public static void tier(long address, int tier) { memory.writeInt(address + TIER_OFFSET, tier); } + + public static long entries(long address) { + return memory.readInt(address + ENTRIES_OFFSET) & UNSIGNED_INT_MASK; + } + + public static void entries(long address, long entries) { + if (entries >= (1L << 32)) { + throw new IllegalStateException("tier entries overflow: up to " + UNSIGNED_INT_MASK + + " supported, " + entries + " given"); + } + memory.writeInt(address + ENTRIES_OFFSET, (int) entries); + } + + public static long deleted(long address) { + return memory.readInt(address + DELETED_OFFSET) & UNSIGNED_INT_MASK; + } + + public static void deleted(long address, long deleted) { + if (deleted >= (1L << 32)) { + throw new IllegalStateException("tier deleted entries count overflow: up to " + + UNSIGNED_INT_MASK + " supported, " + deleted + " given"); + } + memory.writeInt(address + DELETED_OFFSET, (int) deleted); + } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java index 6d49c4278..fdae95a00 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -558,17 +558,6 @@ public abstract class VanillaChronicleHash implements HashEntry, ChecksumEntry public void innerRemoveEntryExceptHashLookupUpdate() { s.free(pos, entrySizeInChunks); - s.entries(s.entries() - 1L); s.incrementModCount(); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java index 246846312..a19cf91fc 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java @@ -66,12 +66,20 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { segmentHeader = BigSegmentHeader.INSTANCE; } - public long entries() { - return segmentHeader.size(segmentHeaderAddress); + public long tierEntries() { + if (tier == 0) { + return segmentHeader.entries(segmentHeaderAddress); + } else { + return TierCountersArea.entries(tierCountersAreaAddr()); + } } - public void entries(long size) { - segmentHeader.size(segmentHeaderAddress, size); + public void tierEntries(long tierEntries) { + if (tier == 0) { + segmentHeader.entries(segmentHeaderAddress, tierEntries); + } else { + TierCountersArea.entries(tierCountersAreaAddr(), tierEntries); + } } long lowestPossiblyFreeChunk() { @@ -91,16 +99,30 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { } } - public long deleted() { - return segmentHeader.deleted(segmentHeaderAddress); + public long tierDeleted() { + if (tier == 0) { + return segmentHeader.deleted(segmentHeaderAddress); + } else { + return TierCountersArea.deleted(tierCountersAreaAddr()); + } } - public void deleted(long deleted) { - segmentHeader.deleted(segmentHeaderAddress, deleted); + public void tierDeleted(long tierDeleted) { + if (tier == 0) { + segmentHeader.deleted(segmentHeaderAddress, tierDeleted); + } else { + TierCountersArea.deleted(tierCountersAreaAddr(), tierDeleted); + } } public long size() { - return entries() - deleted(); + goToFirstTier(); + long size = tierEntries() - tierDeleted(); + while (hasNextTier()) { + nextTier(); + size += tierEntries() - tierDeleted(); + } + return size; } @@ -624,6 +646,7 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { } return -1; } else { + tierEntries(tierEntries() + 1); // if bit at lowestPossiblyFreeChunk is clear, it was skipped because // more than 1 chunk was requested. Don't move lowestPossiblyFreeChunk // in this case. chunks == 1 clause is just a fast path. @@ -647,11 +670,20 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { @Stage("Segment") public void free(long fromPos, int chunks) { + tierEntries(tierEntries() - 1); freeList.clearRange(fromPos, fromPos + chunks); if (fromPos < lowestPossiblyFreeChunk()) lowestPossiblyFreeChunk(fromPos); } + @Stage("Segment") + public void freeExtra(long pos, int oldChunks, int newChunks) { + long from = pos + newChunks; + freeList.clearRange(from, pos + oldChunks); + if (from < lowestPossiblyFreeChunk()) + lowestPossiblyFreeChunk(from); + } + public void verifyTierCountersAreaData() { goToFirstTier(); while (true) { diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java index efb90ed71..9416ece5a 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java @@ -47,6 +47,10 @@ public abstract class HashSegmentIteration> return this; } + public long tierEntriesForIteration() { + return s.tierEntries(); + } + public boolean entryRemovedOnThisIteration = false; abstract boolean entryRemovedOnThisIterationInit(); @@ -69,30 +73,22 @@ public abstract class HashSegmentIteration> public boolean forEachSegmentEntryWhile(Predicate predicate) { checkOnEachPublicOperation.checkOnEachPublicOperation(); s.innerUpdateLock.lock(); - return innerForEachSegmentEntryWhile(predicate, s.size()); + return innerForEachSegmentEntryWhile(predicate); } - public boolean innerForEachSegmentEntryWhile( - Predicate predicate, long expectedEntriesToIterate) { + public boolean innerForEachSegmentEntryWhile(Predicate predicate) { try { - long leftEntries = expectedEntriesToIterate; - if (leftEntries == 0) - return true; s.goToLastTier(); while (true) { int currentTier = s.tier; long currentTierBaseAddr = s.tierBaseAddr; long currentTierIndex = s.tierIndex; - leftEntries = forEachTierWhile(predicate, leftEntries, - currentTier, currentTierBaseAddr, currentTierIndex); - if (leftEntries < 0) // interrupted + boolean interrupted = forEachTierEntryWhile( + predicate, currentTier, currentTierBaseAddr, currentTierIndex); + if (interrupted) return false; - if (leftEntries == 0) - return true; if (currentTier == 0) - throw new IllegalStateException("We iterated all tiers without interruption, " + - "but according to segment counters there should be " + leftEntries + - " more entries. Size diverged?"); + return true; s.prevTier(); } } finally { @@ -102,9 +98,10 @@ public abstract class HashSegmentIteration> } } - private long forEachTierWhile( - Predicate predicate, long leftEntries, + private boolean forEachTierEntryWhile( + Predicate predicate, int currentTier, long currentTierBaseAddr, long tierIndex) { + long leftEntries = tierEntriesForIteration(); boolean interrupted = false; long startPos = 0L; CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup; @@ -160,7 +157,12 @@ public abstract class HashSegmentIteration> // becomes equal to start pos without making the whole loop, but only visiting a single // entry } while (currentHashLookupPos != startPos || steps == 0); - return interrupted ? ~leftEntries : leftEntries; + if (!interrupted && leftEntries > 0) { + throw new IllegalStateException("We a tier without interruption, " + + "but according to tier counters there should be " + leftEntries + + " more entries. Size diverged?"); + } + return interrupted; } @Override diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java index 8b3844dce..2944527de 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java @@ -27,22 +27,11 @@ public class SearchAllocatedChunks extends AllocatedChunks { @StageRef KeySearch ks; - @Override - public void incrementSegmentEntriesIfNeeded() { - if (ks.searchState != PRESENT) { - // update the size before the store fence - s.entries(s.entries() + 1L); - } - } - /** - * @return {@code true} is tier has changed + * @return {@code true} if tier has changed */ public boolean initEntryAndKey(long entrySize) { initAllocatedChunks(hh.h().inChunks(entrySize)); - // call incrementSegmentEntriesIfNeeded() before entry.writeNewEntry(), because the latter - // clears out searchState, and it performs the search again, but in inconsistent state - incrementSegmentEntriesIfNeeded(); int tierBeforeAllocation = s.tier; long pos = alloc.alloc(allocatedChunks, -1, 0); entry.writeNewEntry(pos, ks.inputKey); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java index b3f799ca3..6ec4c46b4 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java @@ -171,8 +171,7 @@ public abstract class MapEntryStages extends HashEntryStages relocation(newValue, newSizeOfEverythingBeforeValue); return; } else if (newSizeInChunks < entrySizeInChunks) { - // Freeing extra chunks - s.free(pos + newSizeInChunks, entrySizeInChunks - newSizeInChunks); + s.freeExtra(pos, entrySizeInChunks, newSizeInChunks); } // new size != old size => size is not constant => size is actually written => // to prevent (at least) this execution: @@ -268,7 +267,7 @@ public abstract class MapEntryStages extends HashEntryStages // fast path if (!mh.m().constantlySizedEntry && mh.m().couldNotDetermineAlignmentBeforeAllocation && entrySizeInChunks < allocatedChunks.allocatedChunks) { - s.free(pos + entrySizeInChunks, allocatedChunks.allocatedChunks - entrySizeInChunks); + s.freeExtra(pos, allocatedChunks.allocatedChunks, entrySizeInChunks); } else { initEntrySizeInChunks(allocatedChunks.allocatedChunks); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java index 2c585c453..9076f965b 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java @@ -66,6 +66,11 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI return !e.entryDeleted() ? entryDelegating : absentEntryDelegating; } + @Override + public long tierEntriesForIteration() { + return entriesToTest == ALL ? s.tierEntries() : s.tierEntries() - s.tierDeleted(); + } + @Override public void doReplaceValue(Data newValue) { checkOnEachPublicOperation.checkOnEachPublicOperation(); @@ -83,7 +88,7 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI checkOnEachPublicOperation.checkOnEachPublicOperation(); initEntriesToTest(PRESENT); s.innerUpdateLock.lock(); - return innerForEachSegmentEntryWhile(predicate, s.size()); + return innerForEachSegmentEntryWhile(predicate); } @Override @@ -92,7 +97,7 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI checkOnEachPublicOperation.checkOnEachPublicOperation(); initEntriesToTest(ALL); s.innerUpdateLock.lock(); - return innerForEachSegmentEntryWhile(predicate, s.entries()); + return innerForEachSegmentEntryWhile(predicate); } @Override @@ -112,7 +117,7 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI e.updatedReplicationStateOnPresentEntry(); e.writeEntryDeleted(); ru.updateChange(); - s.deleted(s.deleted() + 1); + s.tierDeleted(s.tierDeleted() + 1); } finally { s.innerWriteLock.unlock(); } @@ -124,7 +129,7 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI boolean wasDeleted = e.entryDeleted(); super.doRemove(); if (wasDeleted) - s.deleted(s.deleted() - 1); + s.tierDeleted(s.tierDeleted() - 1); } @Override @@ -133,7 +138,7 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI if (e.entryDeleted()) { try { e.innerDefaultReplaceValue(value); - s.deleted(s.deleted() - 1); + s.tierDeleted(s.tierDeleted() - 1); s.incrementModCount(); e.writeEntryPresent(); ru.updateChange(); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/QueryMapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/QueryMapEntryStages.java index f6b416c55..9d094a1c7 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/QueryMapEntryStages.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/QueryMapEntryStages.java @@ -48,7 +48,7 @@ public abstract class QueryMapEntryStages extends MapEntryStages { } if (s.realloc(pos, 0, newSizeInChunks)) { s.innerWriteLock.lock(); - allocatedChunks.incrementSegmentEntriesIfNeeded(); + s.tierEntries(s.tierEntries() + 1); if (newValueSizeIsDifferent) { initValue(newValue); } else { diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java index f050d79f6..ac65ed2fe 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java @@ -46,7 +46,7 @@ public abstract class ReplicatedMapAbsent extends MapAbsent { } else { e.innerDefaultReplaceValue(value); e.updatedReplicationStateOnPresentEntry(); - s.deleted(s.deleted() - 1); + s.tierDeleted(s.tierDeleted() - 1); } s.incrementModCount(); e.writeEntryPresent(); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java index be1b37547..612911432 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java @@ -78,7 +78,7 @@ public abstract class ReplicatedMapQuery extends MapQuery e.updatedReplicationStateOnPresentEntry(); e.writeEntryDeleted(); ru.updateChange(); - s.deleted(s.deleted() + 1); + s.tierDeleted(s.tierDeleted() + 1); } else { throw new IllegalStateException("Entry is absent in the map when doRemove() is called"); } @@ -89,7 +89,7 @@ public abstract class ReplicatedMapQuery extends MapQuery boolean wasDeleted = e.entryDeleted(); super.doRemove(); if (wasDeleted) - s.deleted(s.deleted() - 1L); + s.tierDeleted(s.tierDeleted() - 1L); } @Override diff --git a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java index 72bf3777d..568553c2a 100644 --- a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java @@ -265,7 +265,12 @@ public class EntryCountMapTest { private void dumpMapStats(int segments, int minSize, ChronicleMap map) { - long[] a = ((VanillaChronicleMap) map).segmentSizes(); + long[] a = new long[map.segments()]; + for (int i = 0; i < map.segments(); i++) { + try (MapSegmentContext c = map.segmentContext(i)) { + a[i] = c.size(); + } + } System.out.println("segs: " + segments + " min: " + minSize + " was " + map.size() + " " + Arrays.toString(a) + " sum: " + sum(a));