Make entries and deleted counters per-tier

This commit is contained in:
leventov
2015-12-07 11:43:37 +03:00
parent 2f58bfb399
commit ce45903af7
15 changed files with 125 additions and 85 deletions
@@ -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
@@ -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);
@@ -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);
}
}
@@ -558,17 +558,6 @@ public abstract class VanillaChronicleHash<K,
}
}
/**
* For testing
*/
public final long[] segmentSizes() {
long[] sizes = new long[actualSegments];
for (int i = 0; i < actualSegments; i++) {
sizes[i] = BigSegmentHeader.INSTANCE.size(segmentHeaderAddress(i));
}
return sizes;
}
public final long segmentHeaderAddress(int segmentIndex) {
return bsAddress() + segmentHeadersOffset + ((long) segmentIndex) * segmentHeaderSize;
}
@@ -599,10 +588,10 @@ public abstract class VanillaChronicleHash<K,
@Override
public final long longSize() {
long result = 0L;
for (int i = 0; i < actualSegments; i++) {
long segmentHeaderAddress = segmentHeaderAddress(i);
result += BigSegmentHeader.INSTANCE.size(segmentHeaderAddress) -
BigSegmentHeader.INSTANCE.deleted(segmentHeaderAddress);
for (int i = 0; i < segments(); i++) {
try (SC c = segmentContext(i)) {
result += c.size();
}
}
return result;
}
@@ -33,10 +33,6 @@ public class AllocatedChunks {
public void initAllocatedChunks(int allocatedChunks) {
this.allocatedChunks = allocatedChunks;
}
public void incrementSegmentEntriesIfNeeded() {
// don't increment
}
/**
* @return {@code true} is tier has changed
@@ -44,10 +40,6 @@ public class AllocatedChunks {
public boolean initEntryAndKeyCopying(
long entrySize, long bytesToCopy, long prevPos, int prevChunks) {
initAllocatedChunks(hh.h().inChunks(entrySize));
// call incrementSegmentEntriesIfNeeded() before entry.copyExistingEntry(), because
// the latter clears out searchState, and it performs the search again, but in inconsistent
// state
incrementSegmentEntriesIfNeeded();
long oldSegmentTierBaseAddr = s.tierBaseAddr;
long oldKeySizeAddr = oldSegmentTierBaseAddr + entry.keySizeOffset;
long oldKeyAddr = oldSegmentTierBaseAddr + entry.keyOffset;
@@ -180,7 +180,6 @@ public abstract class HashEntryStages<K> implements HashEntry<K>, ChecksumEntry
public void innerRemoveEntryExceptHashLookupUpdate() {
s.free(pos, entrySizeInChunks);
s.entries(s.entries() - 1L);
s.incrementModCount();
}
}
@@ -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) {
@@ -47,6 +47,10 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
return this;
}
public long tierEntriesForIteration() {
return s.tierEntries();
}
public boolean entryRemovedOnThisIteration = false;
abstract boolean entryRemovedOnThisIterationInit();
@@ -69,30 +73,22 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
public boolean forEachSegmentEntryWhile(Predicate<? super E> predicate) {
checkOnEachPublicOperation.checkOnEachPublicOperation();
s.innerUpdateLock.lock();
return innerForEachSegmentEntryWhile(predicate, s.size());
return innerForEachSegmentEntryWhile(predicate);
}
public <T> boolean innerForEachSegmentEntryWhile(
Predicate<? super T> predicate, long expectedEntriesToIterate) {
public <T> boolean innerForEachSegmentEntryWhile(Predicate<? super T> 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<K, E extends HashEntry<K>>
}
}
private <T> long forEachTierWhile(
Predicate<? super T> predicate, long leftEntries,
private <T> boolean forEachTierEntryWhile(
Predicate<? super T> 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<K, E extends HashEntry<K>>
// 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
@@ -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);
@@ -171,8 +171,7 @@ public abstract class MapEntryStages<K, V> extends HashEntryStages<K>
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<K, V> extends HashEntryStages<K>
// 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);
}
@@ -66,6 +66,11 @@ public abstract class ReplicatedMapSegmentIteration<K, V, R> 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<V> newValue) {
checkOnEachPublicOperation.checkOnEachPublicOperation();
@@ -83,7 +88,7 @@ public abstract class ReplicatedMapSegmentIteration<K, V, R> 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<K, V, R> 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<K, V, R> 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<K, V, R> 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<K, V, R> 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();
@@ -48,7 +48,7 @@ public abstract class QueryMapEntryStages<K, V> extends MapEntryStages<K, V> {
}
if (s.realloc(pos, 0, newSizeInChunks)) {
s.innerWriteLock.lock();
allocatedChunks.incrementSegmentEntriesIfNeeded();
s.tierEntries(s.tierEntries() + 1);
if (newValueSizeIsDifferent) {
initValue(newValue);
} else {
@@ -46,7 +46,7 @@ public abstract class ReplicatedMapAbsent<K, V> extends MapAbsent<K, V> {
} else {
e.innerDefaultReplaceValue(value);
e.updatedReplicationStateOnPresentEntry();
s.deleted(s.deleted() - 1);
s.tierDeleted(s.tierDeleted() - 1);
}
s.incrementModCount();
e.writeEntryPresent();
@@ -78,7 +78,7 @@ public abstract class ReplicatedMapQuery<K, V, R> extends MapQuery<K, V, R>
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<K, V, R> extends MapQuery<K, V, R>
boolean wasDeleted = e.entryDeleted();
super.doRemove();
if (wasDeleted)
s.deleted(s.deleted() - 1L);
s.tierDeleted(s.tierDeleted() - 1L);
}
@Override