diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index 34d92bad1..21daf2b7a 100755 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -230,24 +230,69 @@ public interface ChronicleHashBuilder, B maxChunksPerEntry(int maxChunksPerEntry); /** - * Configures the maximum number of entries, that could be inserted into the hash containers, - * created by this builder. If you try to insert more data, {@link IllegalStateException} - * might be thrown, because currently {@link ChronicleMap} and {@link ChronicleSet} - * don't support resizing. + * Configures the target number of entries, that is going be inserted into the hash containers, + * created by this builder. If {@link #maxBloatFactor(double)} is configured to {code 1.0} + * (and this is by default), this number of entries is also the maximum. If you try to insert + * more entries, than the configured {@code maxBloatFactor}, multiplied by the given number of + * {@code entries}, {@link IllegalStateException} might be thrown. * - *

You shouldn't put additional margin over the actual maximum number of entries. + *

This configuration should represent the expected maximum number of entries in a stable + * state, {@link #maxBloatFactor(double) maxBloatFactor} - the maximum bloat up coefficient, + * during exceptional bursts. + * + *

To be more precise - try to configure the {@code entries} so, that the created hash + * container is going to serve about 99% requests being less or equal than this number + * of entries in size. + * + *

You shouldn't put additional margin over the actual target number of entries. * This bad practice was popularized by {@link HashMap#HashMap(int)} and {@link * HashSet#HashSet(int)} constructors, which accept capacity, that should be multiplied * by load factor to obtain the actual maximum expected number of entries. * {@code ChronicleMap} and {@code ChronicleSet} don't have a notion of load factor. * - *

Default maximum entries is 2^20 (~ 1 million). + *

The default target number of entries is 2^20 (~ 1 million). * - * @param entries maximum size of the maps or sets, created by this builder + * @param entries the target size of the maps or sets, created by this builder * @return this builder back + * @throws IllegalArgumentException if the given {@code entries} number is non-positive + * @see #maxBloatFactor(double) */ B entries(long entries); + /** + * Configures the maximum number of times, the hash containers, created by this builder, + * are allowed to grow in size beyond the configured {@linkplain #entries(long) target number + * of entries}. + * + *

{@link #entries(long)} should represent the expected maximum number of entries in a stable + * state, {@code maxBloatFactor} - the maximum bloat up coefficient, during exceptional bursts. + * + *

This configuration should be used for self-checking. Even if you configure impossibly + * large {@code maxBloatFactor}, the created {@code ChronicleHash}, of cause, will be still + * operational, and even won't allocate any extra resources before they are actually needed. + * But when the {@code ChronicleHash} grows beyond the configured {@link #entries(long)}, it + * could start to serve requests progressively slower. If you insert new entries into + * {@code ChronicleHash} infinitely, due to a bug in your business logic code, or the + * ChronicleHash configuration, and if you configure the ChronicleHash to grow infinitely, you + * will have a terribly slow and fat, but operational application, instead of a fail with + * {@code IllegalStateException}, which will quickly show you, that there is a bug in you + * application. + * + *

The default maximum bloat factor factor is {@code 1.0} - i. e. "no bloat is expected". + * + *

It is strongly advised not to configure {@code maxBloatFactor} to more than {@code 10.0}, + * almost certainly, you either should configure {@code ChronicleHash}es completely differently, + * or this data structure doesn't fit you case. + * + * @param maxBloatFactor the maximum number ot times, the created hash container is supposed + * to bloat up beyond the {@link #entries(long)} + * @return this builder back + * @throws IllegalArgumentException if the given {@code maxBloatFactor} is NaN, lesser than 1.0 + * or greater than 1000.0 (one thousand) + * @see #entries(long) + */ + B maxBloatFactor(double maxBloatFactor); + /** * Configures the actual maximum number entries, that could be inserted into any single segment * of the hash containers, created by this builder. Configuring both the actual number of diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilderPrivateAPI.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilderPrivateAPI.java index eac41a5a3..56eef896e 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilderPrivateAPI.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilderPrivateAPI.java @@ -35,4 +35,6 @@ public interface ChronicleHashBuilderPrivateAPI { int segmentHeaderSize(boolean replicated); int actualSegments(boolean replicated); + + long maxExtraTiers(boolean replicated); } 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 67e35bd57..0630d53f2 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java @@ -66,6 +66,10 @@ public final class BigSegmentHeader implements SegmentHeader { @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"); + } UNSAFE.putInt(address + SIZE_OFFSET, (int) size); } @@ -76,6 +80,10 @@ public final class BigSegmentHeader implements SegmentHeader { @Override public void deleted(long address, long deleted) { + if (deleted >= (1L << 32)) { + throw new IllegalStateException("segment deleted entries count overflow: up to " + + UNSIGNED_INT_MASK + " supported, " + deleted + " given"); + } UNSAFE.putInt(address + DELETED_OFFSET, (int) deleted); } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/TierData.java b/src/main/java/net/openhft/chronicle/hash/impl/TierData.java new file mode 100644 index 000000000..32dc43fce --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/TierData.java @@ -0,0 +1,70 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.hash.impl; + +import static net.openhft.lang.io.NativeBytes.UNSAFE; + +public enum TierData { + ; + + 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 NEXT_POS_TO_SEARCH_FROM_TIERED_OFFSET = PREV_TIER_INDEX_OFFSET + 8L; + public static final long SEGMENT_INDEX_OFFSET = NEXT_POS_TO_SEARCH_FROM_TIERED_OFFSET + 8L; + public static final long TIER_OFFSET = SEGMENT_INDEX_OFFSET + 4L; + + public static long nextTierIndex(long address) { + return UNSAFE.getLong(address + NEXT_TIER_INDEX_OFFSET); + } + + public static void nextTierIndex(long address, long nextTierIndex) { + UNSAFE.putLong(address + NEXT_TIER_INDEX_OFFSET, nextTierIndex); + } + + public static long nextPosToSearchFromTiered(long address) { + return UNSAFE.getLong(address + NEXT_POS_TO_SEARCH_FROM_TIERED_OFFSET); + } + + public static void nextPosToSearchFromTiered(long address, long nextPosToSearchFrom) { + UNSAFE.putLong(address + NEXT_POS_TO_SEARCH_FROM_TIERED_OFFSET, + nextPosToSearchFrom); + } + + public static long prevTierIndex(long address) { + return UNSAFE.getLong(address + PREV_TIER_INDEX_OFFSET); + } + + public static void prevTierIndex(long address, long prevTierIndex) { + UNSAFE.putLong(address + PREV_TIER_INDEX_OFFSET, prevTierIndex); + } + + public static int segmentIndex(long address) { + return UNSAFE.getInt(address + SEGMENT_INDEX_OFFSET); + } + + public static void segmentIndex(long address, int segmentIndex) { + UNSAFE.putInt(address + SEGMENT_INDEX_OFFSET, segmentIndex); + } + + public static int tier(long address) { + return UNSAFE.getInt(address + TIER_OFFSET); + } + + public static void tier(long address, int tier) { + UNSAFE.putInt(address + TIER_OFFSET, tier); + } +} 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 2eae066fb..088b4406b 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -16,6 +16,8 @@ package net.openhft.chronicle.hash.impl; +import net.openhft.chronicle.bytes.NativeBytesStore; +import net.openhft.chronicle.core.Maths; import net.openhft.chronicle.hash.*; import net.openhft.chronicle.hash.impl.util.BuildVersion; import net.openhft.chronicle.hash.serialization.BytesReader; @@ -24,10 +26,7 @@ import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.chronicle.hash.serialization.internal.MetaProvider; import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder; import net.openhft.chronicle.map.ChronicleMapBuilder; -import net.openhft.lang.io.Bytes; -import net.openhft.lang.io.BytesStore; -import net.openhft.lang.io.MappedStore; -import net.openhft.lang.io.NativeBytes; +import net.openhft.lang.io.*; import net.openhft.lang.io.serialization.BytesMarshallableSerializer; import net.openhft.lang.threadlocal.Provider; @@ -36,14 +35,14 @@ import java.io.IOException; import java.io.ObjectInputStream; import java.io.Serializable; import java.nio.channels.FileChannel; +import java.util.ArrayList; import java.util.List; import static java.lang.Long.numberOfTrailingZeros; import static java.lang.Math.max; -import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.entrySize; -import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.keyBits; -import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.valueBits; +import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.*; import static net.openhft.lang.MemoryUnit.*; +import static net.openhft.lang.io.NativeBytes.UNSAFE; public abstract class VanillaChronicleHash, C extends HashEntry, SC extends HashSegmentContext, @@ -101,13 +100,44 @@ public abstract class VanillaChronicleHash( + langBytes.address(), langBytes.capacity(), null, false); + this.offset = offset; + } + + public TierBulkData(TierBulkData data, long offset) { + this.langBytesStore = data.langBytesStore; + this.langBytes = data.langBytes; + this.chBytesBytesStore = data.chBytesBytesStore; + this.offset = offset; + } + } + + public transient List tierBulkOffsets; + public transient long headerSize; transient long segmentHeadersOffset; transient long segmentsOffset; @@ -167,12 +197,17 @@ public abstract class VanillaChronicleHash sizeInBytesWithoutTiers() means there are some tiered segments + Math.max(sizeInBytesWithoutTiers(), file.length()), + BytesMarshallableSerializer.create())); // newly-extended file contents are not guaranteed to be zero - if (created) + if (created) { + // TODO zero out only hash lookup tables and bit sets, not entry data areas bytes.zeroOut(headerSize, bytes.capacity(), true); + // zero out tier data + bytes.zeroOut(headerSize, headerSize + 64, true); + } + } + + private void initTierBulks(long byteStoreSize) { + tierBulkOffsets = new ArrayList<>(); + long tierBulkOffset = roundUpRuntimePageSize(sizeInBytesWithoutTiers()); + if (byteStoreSize >= tierBulkOffset + tierBulkBytesSize) { + TierBulkData firstTierBulkData = new TierBulkData(ms, tierBulkOffset); + tierBulkOffsets.add(firstTierBulkData); + while (true) { + tierBulkOffset += tierBulkBytesSize; + if (byteStoreSize < tierBulkOffset + tierBulkBytesSize) + break; + tierBulkOffsets.add(new TierBulkData(firstTierBulkData, tierBulkOffset)); + } + } } private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { @@ -246,12 +323,16 @@ public abstract class VanillaChronicleHash 0) { + tiersInUse(tiersInUse + 1); + long tierBaseAddr = tierIndexToBaseAddr(firstFreeTierIndex); + long tierDataAddr = tierBaseAddr + segmentHashLookupOuterSize; + long nextFreeTierIndex = TierData.nextTierIndex(tierDataAddr); + // now, when this tier will be a part of the map, the next tier designates + // the next tier in the data structure, should be 0 + TierData.nextTierIndex(tierDataAddr, 0); + TierData.segmentIndex(tierDataAddr, forSegmentIndex); + TierData.tier(tierDataAddr, tier); + firstFreeTierIndex(nextFreeTierIndex); + return firstFreeTierIndex; + } else { + allocateTierBulk(); + } + } + } finally { + tierDataUnlock(); + } + } + + private void allocateTierBulk() { + long alreadyAllocatedBulks = extraAllocatedTierBulkCount(); + mapTiers(alreadyAllocatedBulks); + extraAllocatedTierBulkCount(alreadyAllocatedBulks + 1); + linkFreeTiersList(alreadyAllocatedBulks); + } + + private void linkFreeTiersList(long alreadyAllocatedBulks) { + long alreadyAllocatedTiers = alreadyAllocatedBulks * numberOfTiersInBulk; + for (long tierIndex = alreadyAllocatedTiers; + tierIndex < alreadyAllocatedTiers + numberOfTiersInBulk - 1; tierIndex++) { + long tierBaseAddr = tierIndexToBaseAddr(tierIndex); + long tierDataAddr = tierBaseAddr + segmentHashLookupOuterSize; + TierData.nextTierIndex(tierDataAddr, tierIndex + 1); + } + } + + public long tierIndexToBaseAddr(long tierIndex) { + // tiers are 1-counted, to allow tierIndex = 0 to be un-initialized in off-heap memory, + // convert into 0-based form + tierIndex -= 1; + if (tierIndex < actualSegments) + return segmentBaseAddr((int) tierIndex); + return extraTierIndexToBaseAddress(tierIndex); + } + + public Bytes tierBytes(long tierIndex) { + tierIndex -= 1; + if (tierIndex < actualSegments) + return bytes; + return tierBulkData(tierIndex).langBytes; + } + + public long bytesOffset(long tierIndex) { + tierIndex -= 1; + if (tierIndex < actualSegments) + return msBytesOffset(tierIndex); + return tierBulkData(tierIndex).offset; + } + + private TierBulkData tierBulkData(long tierIndex) { + long extraTierIndex = tierIndex - actualSegments; + long bulkIndex = extraTierIndex >> log2NumberOfTiersInBulk; + if (bulkIndex >= tierBulkOffsets.size()) + mapTiers(bulkIndex); + return tierBulkOffsets.get((int) bulkIndex); + } + + private long extraTierIndexToBaseAddress(long tierIndex) { + long extraTierIndex = tierIndex - actualSegments; + long bulkIndex = extraTierIndex >> log2NumberOfTiersInBulk; + if (bulkIndex >= tierBulkOffsets.size()) + mapTiers(bulkIndex); + TierBulkData tierBulkData = tierBulkOffsets.get((int) bulkIndex); + return tierBulkData.langBytes.address() + tierBulkData.offset + + tierBulkInnerOffsetToTiers + + (extraTierIndex & (numberOfTiersInBulk - 1)) * segmentSize; + } + + private void mapTiers(long upToBulkIndex) { + if (ms instanceof MappedStore) { + try { + mapTiersMapped(upToBulkIndex); + } catch (IOException e) { + throw new RuntimeException(e); + } + } else { + // in-memory ChMap + allocateTiers(upToBulkIndex); + } + } + + private void mapTiersMapped(long upToBulkIndex) throws IOException { + int firstBulkToMap = tierBulkOffsets.size(); + long bulksToMap = upToBulkIndex + 1 - firstBulkToMap; + long mapSize = bulksToMap * tierBulkBytesSize; + long mapStart = roundUpRuntimePageSize(sizeInBytesWithoutTiers()) + + firstBulkToMap * tierBulkBytesSize; + MappedStore extraStore = new MappedStore(file(), FileChannel.MapMode.READ_WRITE, + mapStart, mapSize, ms.objectSerializer()); + appendTierBulkData(upToBulkIndex, firstBulkToMap, extraStore); + } + + private void allocateTiers(long upToBulkIndex) { + int firstBulkToMap = tierBulkOffsets.size(); + long bulksToMap = upToBulkIndex + 1 - firstBulkToMap; + long mapSize = bulksToMap * tierBulkBytesSize; + DirectStore extraStore = new DirectStore(ms.objectSerializer(), mapSize, true); + appendTierBulkData(upToBulkIndex, firstBulkToMap, extraStore); + } + + private void appendTierBulkData(long upToBulkIndex, int firstBulkToMap, BytesStore extraStore) { + long offset = 0; + TierBulkData firstMappedTierBulkData = new TierBulkData(extraStore, offset); + tierBulkOffsets.add(firstMappedTierBulkData); + for (long i = firstBulkToMap + 1; i <= upToBulkIndex; i++) { + tierBulkOffsets.add(new TierBulkData(firstMappedTierBulkData, + offset += tierBulkBytesSize)); + } + } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/EntryKeyBytesData.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/EntryKeyBytesData.java index fd722af28..7b00d8376 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/EntryKeyBytesData.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/EntryKeyBytesData.java @@ -19,6 +19,7 @@ package net.openhft.chronicle.hash.impl.stage.data.bytes; import net.openhft.chronicle.bytes.RandomDataInput; import net.openhft.chronicle.hash.AbstractData; import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages; +import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop; import net.openhft.sg.Stage; @@ -29,6 +30,7 @@ import net.openhft.sg.Staged; public class EntryKeyBytesData extends AbstractData { @StageRef KeyBytesInterop ki; + @StageRef SegmentStages s; @StageRef HashEntryStages entry; @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation; @@ -44,7 +46,7 @@ public class EntryKeyBytesData extends AbstractData { @Override public RandomDataInput bytes() { checkOnEachPublicOperation.checkOnEachPublicOperation(); - return entry.entryBS; + return s.segmentBS; } @Override @@ -72,7 +74,7 @@ public class EntryKeyBytesData extends AbstractData { } private K innerGetUsing(K usingKey) { - entry.entryBytes.position(entry.keyOffset); - return ki.keyReader.read(entry.entryBytes, size(), usingKey); + s.segmentBytes.position(entry.keyOffset); + return ki.keyReader.read(s.segmentBytes, size(), usingKey); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/AllocatedChunks.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/AllocatedChunks.java index 23ac0eef0..77988a950 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/AllocatedChunks.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/AllocatedChunks.java @@ -37,12 +37,21 @@ public class AllocatedChunks { // don't increment } - public void initEntryAndKeyCopying(long entrySize, long bytesToCopy) { + /** + * @return {@code true} is tier has changed + */ + public boolean initEntryAndKeyCopying(long entrySize, long bytesToCopy) { 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(); - entry.copyExistingEntry(s.alloc(allocatedChunks), bytesToCopy); + long oldSegmentTierBaseAddr = s.segmentBaseAddr; + long oldKeySizeAddr = oldSegmentTierBaseAddr + entry.keySizeOffset; + long oldKeyAddr = oldSegmentTierBaseAddr + entry.keyOffset; + int tierBeforeAllocation = s.segmentTier; + long pos = s.alloc(allocatedChunks); + entry.copyExistingEntry(pos, bytesToCopy, oldKeyAddr, oldKeySizeAddr); + return s.segmentTier > tierBeforeAllocation; } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java index ae31aeaf7..b0fcc2c49 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.hash.impl.stage.entry; +import net.openhft.chronicle.algo.bytes.Access; import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.bytes.NativeBytesStore; import net.openhft.chronicle.hash.Data; @@ -29,6 +30,9 @@ import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import org.jetbrains.annotations.NotNull; +import static net.openhft.chronicle.algo.bytes.Access.checkedBytesStoreAccess; +import static net.openhft.chronicle.algo.bytes.Access.nativeAccess; + @Staged public abstract class HashEntryStages implements HashEntry { @@ -37,10 +41,6 @@ public abstract class HashEntryStages implements HashEntry { @StageRef public SegmentStages s; @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation; @StageRef public HashLookupPos hlp; - - public final Bytes entryBytes = hh.h().ms.bytes(); - public final BytesStore entryBS = - new NativeBytesStore<>(entryBytes.address(), entryBytes.capacity(), null, false); public long pos = -1; @@ -54,7 +54,7 @@ public abstract class HashEntryStages implements HashEntry { public void initEntryOffset() { keySizeOffset = s.entrySpaceOffset + pos * hh.h().chunkSize; - entryBytes.limit(entryBytes.capacity()); + s.segmentBytes.limit(s.segmentBytes.capacity()); } public long keySize = -1; @@ -71,26 +71,32 @@ public abstract class HashEntryStages implements HashEntry { public void readExistingEntry(long pos) { initPos(pos); - entryBytes.position(keySizeOffset); - initKeySize(hh.h().keySizeMarshaller.readSize(entryBytes)); - initKeyOffset(entryBytes.position()); + s.segmentBytes.position(keySizeOffset); + initKeySize(hh.h().keySizeMarshaller.readSize(s.segmentBytes)); + initKeyOffset(s.segmentBytes.position()); } public void writeNewEntry(long pos, Data key) { initPos(pos); initKeySize(key.size()); - entryBytes.position(keySizeOffset); - hh.h().keySizeMarshaller.writeSize(entryBytes, keySize); - initKeyOffset(entryBytes.position()); - key.writeTo(entryBS, keyOffset); + s.segmentBytes.position(keySizeOffset); + hh.h().keySizeMarshaller.writeSize(s.segmentBytes, keySize); + initKeyOffset(s.segmentBytes.position()); + key.writeTo(s.segmentBS, keyOffset); } - public void copyExistingEntry(long newPos, long bytesToCopy) { - long oldKeySizeOffset = keySizeOffset; - long oldKeyOffset = keyOffset; + public void copyExistingEntry( + long newPos, long bytesToCopy, long oldKeyAddr, long oldKeySizeAddr) { initPos(newPos); - initKeyOffset(keySizeOffset + (oldKeyOffset - oldKeySizeOffset)); - entryBS.write(keySizeOffset, entryBS, oldKeySizeOffset, bytesToCopy); + initKeyOffset(keySizeOffset + (oldKeyAddr - oldKeySizeAddr)); + // Calling Access.copy() which is probably slower because not of abstractions, + // because there is no BytesStore.write(off, addr, len) method. Alternative is + // to make a final BytesStore rawMemoryStore = new PointerBytesStore().set(0, Long.MAX_V) + // and here: s.segmentBS.write(keySizeOffset, rawMemoryStore, keySizeAddr, bytesToCopy) + Access.copy( + nativeAccess(), null, oldKeySizeAddr, + checkedBytesStoreAccess(), s.segmentBS, keySizeOffset, + bytesToCopy); } public long keyEnd() { diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java index 6f4fab74c..db67e0266 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java @@ -46,6 +46,6 @@ public abstract class HashLookupPos { public void putValueVolatile(long newValue) { CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup; hashLookup.checkValueForPut(newValue); - hashLookup.putValueVolatile(s.segmentBase, hashLookupPos, newValue); + hashLookup.putValueVolatile(s.segmentBaseAddr, hashLookupPos, newValue); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashLookupSearch.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupSearch.java similarity index 81% rename from src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashLookupSearch.java rename to src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupSearch.java index 129613643..4d03a9554 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashLookupSearch.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupSearch.java @@ -14,13 +14,15 @@ * along with this program. If not, see . */ -package net.openhft.chronicle.hash.impl.stage.query; +package net.openhft.chronicle.hash.impl.stage.entry; import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages; import net.openhft.chronicle.hash.impl.stage.entry.HashLookupPos; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; +import net.openhft.chronicle.hash.impl.stage.query.HashQuery; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.sg.Stage; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -31,26 +33,24 @@ import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.UNSE public abstract class HashLookupSearch { @StageRef SegmentStages s; - @StageRef HashEntryStages e; - @StageRef HashQuery op; - @StageRef VanillaChronicleHashHolder hh; + @StageRef public VanillaChronicleHashHolder hh; @StageRef HashLookupPos hlp; @StageRef KeySearch ks; @Stage("SearchKey") long searchKey = UNSET_KEY; - @Stage("SearchKey") long searchStartPos; + @Stage("SearchKey") public long searchStartPos; - private CompactOffHeapLinearHashTable hl() { + public CompactOffHeapLinearHashTable hl() { return hh.h().hashLookup; } - - void initSearchKey() { - searchKey = hl().maskUnsetKey(hh.h().hashSplitting.segmentHash(op.hashOfKey)); + + public void initSearchKey(long searchKey) { + this.searchKey = searchKey; searchStartPos = hl().hlPos(searchKey); } private long addr() { - return s.segmentBase; + return s.segmentBaseAddr; } public long nextPos() { @@ -83,6 +83,10 @@ public abstract class HashLookupSearch { } public void putNewVolatile(long value) { + // correctness check + make putNewVolatile() dependant on keySearch, this, in turn, + // is needed for hlp.hashLookupPos re-initialization after nextTier() + assert !ks.searchStatePresent(); + hl().checkValueForPut(value); long currentEntry = hl().readEntry(addr(), hlp.hashLookupPos); hl().writeEntryVolatile(addr(), hlp.hashLookupPos, currentEntry, searchKey, value); 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 ebce419a1..348bb5330 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 @@ -16,6 +16,10 @@ package net.openhft.chronicle.hash.impl.stage.entry; +import net.openhft.chronicle.algo.bitset.ReusableBitSet; +import net.openhft.chronicle.algo.bitset.SingleThreadedFlatBitSetFrame; +import net.openhft.chronicle.algo.bytes.Access; +import net.openhft.chronicle.bytes.PointerBytesStore; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.SegmentLock; import net.openhft.chronicle.hash.impl.*; @@ -23,9 +27,9 @@ import net.openhft.chronicle.hash.impl.stage.hash.Chaining; import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.hash.locks.InterProcessLock; +import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.lang.collection.DirectBitSet; -import net.openhft.lang.collection.SingleThreadedDirectBitSet; -import net.openhft.lang.io.MultiStoreBytes; +import net.openhft.lang.io.Bytes; import net.openhft.sg.Stage; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -33,6 +37,8 @@ import org.jetbrains.annotations.NotNull; import java.util.Objects; +import static net.openhft.chronicle.algo.MemoryUnit.BITS; +import static net.openhft.chronicle.algo.MemoryUnit.LONGS; import static net.openhft.chronicle.hash.impl.LocalLockState.UNLOCKED; @Staged @@ -54,7 +60,7 @@ public abstract class SegmentStages implements SegmentLock { abstract boolean segmentHeaderInit(); private void initSegmentHeader() { - segmentHeaderAddress = hh.h().ms.address() + hh.h().segmentHeaderOffset(segmentIndex); + segmentHeaderAddress = hh.h().segmentHeaderAddress(segmentIndex); segmentHeader = BigSegmentHeader.INSTANCE; } @@ -67,11 +73,19 @@ public abstract class SegmentStages implements SegmentLock { } long nextPosToSearchFrom() { - return segmentHeader.nextPosToSearchFrom(segmentHeaderAddress); + if (segmentTier == 0) { + return segmentHeader.nextPosToSearchFrom(segmentHeaderAddress); + } else { + return nextPosToSearchFromTiered(); + } } public void nextPosToSearchFrom(long nextPosToSearchFrom) { - segmentHeader.nextPosToSearchFrom(segmentHeaderAddress, nextPosToSearchFrom); + if (segmentTier == 0) { + segmentHeader.nextPosToSearchFrom(segmentHeaderAddress, nextPosToSearchFrom); + } else { + nextPosToSearchFromTiered(nextPosToSearchFrom); + } } public long deleted() { @@ -346,11 +360,91 @@ public abstract class SegmentStages implements SegmentLock { checkOnEachPublicOperation.checkOnEachPublicOperation(); return innerWriteLock; } + + @Stage("SegmentTier") public int segmentTier = -1; + @Stage("SegmentTier") public long tierIndex; + @Stage("SegmentTier") public long segmentBaseAddr; + + public void initSegmentTier() { + tierIndex = segmentIndex + 1; // tiers are 1-counted + segmentBaseAddr = hh.h().segmentBaseAddr(segmentIndex); + // assign the field of stage on which init() checks last, because flushed in segmentIndex() + // initialization + // TODO this is dangerous... review other stages and compilation mechanism + segmentTier = 0; + } + + private void initSegmentTier(int tier, long tierIndex) { + segmentTier = tier; + this.tierIndex = tierIndex; + assert tierIndex > 0; + this.segmentBaseAddr = hh.h().tierIndexToBaseAddr(tierIndex); + } + + public void initSegmentTier(int tier, long tierIndex, long tierBaseAddr) { + segmentTier = tier; + this.tierIndex = tierIndex; + this.segmentBaseAddr = tierBaseAddr; + } + + public long tierDataOffset() { + return segmentBaseAddr + hh.h().segmentHashLookupOuterSize; + } + + public long nextTierIndex() { + return TierData.nextTierIndex(tierDataOffset()); + } + + public void nextTierIndex(long nextTierIndex) { + TierData.nextTierIndex(tierDataOffset(), nextTierIndex); + } + + public long nextPosToSearchFromTiered() { + return TierData.nextPosToSearchFromTiered(tierDataOffset()); + } + + public void nextPosToSearchFromTiered(long nextPosToSearchFrom) { + TierData.nextPosToSearchFromTiered(tierDataOffset(), nextPosToSearchFrom); + } + + public long prevTierIndex() { + return TierData.prevTierIndex(tierDataOffset()); + } + + public void prevTierIndex(long prevTierIndex) { + TierData.prevTierIndex(tierDataOffset(), prevTierIndex); + } + + public void nextTier() { + VanillaChronicleHash, ?, ?, ?> h = hh.h(); + long nextTierIndex = nextTierIndex(); + if (nextTierIndex == 0) { + nextTierIndex = h.allocateTier(segmentIndex, segmentTier + 1); + nextTierIndex(nextTierIndex); + long currentTierBase = segmentBaseAddr; + initSegmentTier(segmentTier + 1, nextTierIndex); + prevTierIndex(currentTierBase); + } else { + initSegmentTier(segmentTier + 1, nextTierIndex); + } + } + + public boolean hasNextTier() { + return nextTierIndex() != 0; + } + + public void prevTier() { + if (segmentTier == 0) + throw new IllegalStateException("first tier doesn't have previous"); + initSegmentTier(segmentTier - 1, prevTierIndex()); + } - @Stage("Segment") MultiStoreBytes freeListBytes = new MultiStoreBytes(); - @Stage("Segment") public SingleThreadedDirectBitSet freeList = new SingleThreadedDirectBitSet(); + @Stage("Segment") public final PublicMultiStoreBytes segmentBytes = new PublicMultiStoreBytes(); + @Stage("Segment") public final PointerBytesStore segmentBS = new PointerBytesStore(); + @Stage("Segment") public final ReusableBitSet freeList = new ReusableBitSet( + new SingleThreadedFlatBitSetFrame(LONGS.align(hh.h().actualChunksPerSegment, BITS)), + Access.nativeAccess(), null, 0); @Stage("Segment") long entrySpaceOffset = 0; - @Stage("Segment") public long segmentBase; boolean segmentInit() { return entrySpaceOffset > 0; @@ -358,11 +452,17 @@ public abstract class SegmentStages implements SegmentLock { void initSegment() { VanillaChronicleHash h = hh.h(); - long hashLookupOffset = h.segmentOffset(segmentIndex); - segmentBase = hh.h().ms.address() + hashLookupOffset; - long freeListOffset = hashLookupOffset + h.segmentHashLookupOuterSize; - freeListBytes.storePositionAndSize(h.ms, freeListOffset, h.segmentFreeListInnerSize); - freeList.reuse(freeListBytes); + + PublicMultiStoreBytes segmentBytes = this.segmentBytes; + segmentBytes.setBytesOffset(h.tierBytes(tierIndex), h.bytesOffset(tierIndex)); + + long segmentBase = this.segmentBaseAddr; + segmentBS.set(segmentBase, h.segmentSize); + + // 64 is a cache line with tier data + long freeListOffset = h.segmentHashLookupOuterSize + 64; + freeList.setOffset(segmentBase + freeListOffset); + entrySpaceOffset = freeListOffset + h.segmentFreeListOuterSize + h.segmentEntrySpaceInnerOffset; } @@ -373,21 +473,22 @@ public abstract class SegmentStages implements SegmentLock { public long allocReturnCode(int chunks) { VanillaChronicleHash h = hh.h(); - if (chunks > h.maxChunksPerEntry) + if (chunks > h.maxChunksPerEntry) { throw new IllegalArgumentException("Entry is too large: requires " + chunks + " entry size chucks, " + h.maxChunksPerEntry + " is maximum."); + } long ret = freeList.setNextNContinuousClearBits(nextPosToSearchFrom(), chunks); if (ret == DirectBitSet.NOT_FOUND || ret + chunks > h.actualChunksPerSegment) { if (ret != DirectBitSet.NOT_FOUND && ret + chunks > h.actualChunksPerSegment && ret < h.actualChunksPerSegment) { - freeList.clear(ret, h.actualChunksPerSegment); + freeList.clearRange(ret, h.actualChunksPerSegment); } ret = freeList.setNextNContinuousClearBits(0L, chunks); if (ret == DirectBitSet.NOT_FOUND || ret + chunks > h.actualChunksPerSegment) { if (ret != DirectBitSet.NOT_FOUND && ret + chunks > h.actualChunksPerSegment && ret < h.actualChunksPerSegment) { - freeList.clear(ret, h.actualChunksPerSegment); + freeList.clearRange(ret, h.actualChunksPerSegment); } return -1; } @@ -404,24 +505,18 @@ public abstract class SegmentStages implements SegmentLock { } public long alloc(int chunks) { - long ret = allocReturnCode(chunks); - if (ret >= 0) { - return ret; - } else { - if (chunks == 1) { - throw new IllegalStateException( - "Segment is full, no free entries found"); + while (true) { + long ret = allocReturnCode(chunks); + if (ret >= 0) { + return ret; } else { - throw new IllegalStateException( - "Segment is full or has no ranges of " + chunks - + " continuous free chunks" - ); + nextTier(); } } } public void free(long fromPos, int chunks) { - freeList.clear(fromPos, fromPos + chunks); + freeList.clearRange(fromPos, fromPos + chunks); if (fromPos < nextPosToSearchFrom()) nextPosToSearchFrom(fromPos); } @@ -432,16 +527,4 @@ public abstract class SegmentStages implements SegmentLock { nextPosToSearchFrom = 0L; nextPosToSearchFrom(nextPosToSearchFrom); } - - public void clearSegment() { - innerWriteLock.lock(); - hh.h().hashLookup.clearHashLookup(segmentBase); - freeList.clear(); - nextPosToSearchFrom(0L); - entries(0L); - } - - public void clear() { - clearSegment(); - } } 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 f6dd85e6d..65af1d8e9 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 @@ -34,7 +34,7 @@ import java.util.function.Predicate; public abstract class HashSegmentIteration> implements HashEntry, HashSegmentContext { - @StageRef public SegmentStages s; + @StageRef public IterationSegmentStages s; @StageRef HashEntryStages e; @StageRef VanillaChronicleHashHolder hh; @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation; @@ -52,6 +52,20 @@ public abstract class HashSegmentIteration> this.entryRemovedOnThisIteration = entryRemovedOnThisIteration; } + public long hashLookupEntry = 0; + + public void initHashLookupEntry(long entry) { + hashLookupEntry = entry; + } + + abstract void closeHashLookupEntry(); + + private void goToLastTier() { + while (s.hasNextTier()) { + s.nextTier(); + } + } + @Override public boolean forEachSegmentEntryWhile(Predicate action) { s.innerUpdateLock.lock(); @@ -59,20 +73,46 @@ public abstract class HashSegmentIteration> long size = s.size(); if (size == 0) return true; - boolean interrupted = false; - long startPos = 0L; - CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup; - while (!hashLookup.empty(hashLookup.readEntry(s.segmentBase, startPos))) { - startPos = hashLookup.step(startPos); + goToLastTier(); + while (true) { + int currentTier = s.segmentTier; + long currentTierBaseAddr = s.segmentBaseAddr; + size = forEachTierWhile(action, size, currentTier, currentTierBaseAddr); + if (size < 0) // interrupted + return false; + if (size == 0) + return true; + if (currentTier == 0) + throw new IllegalStateException("We iterated all tiers without interruption, " + + "but according to size counter there should be " + size + " more " + + "entries. Size diverged?"); + s.prevTier(); } - hlp.initHashLookupPos(startPos); - do { - hlp.setHashLookupPos(hashLookup.step(hlp.hashLookupPos)); - long entry = hashLookup.readEntry(s.segmentBase, hlp.hashLookupPos); - if (!hashLookup.empty(entry)) { - e.readExistingEntry(hashLookup.value(entry)); - if (entryIsPresent()) { - initEntryRemovedOnThisIteration(false); + } finally { + closeHashLookupEntry(); + s.innerReadLock.unlock(); + initEntryRemovedOnThisIteration(false); + } + } + + private long forEachTierWhile( + Predicate action, long size, int currentTier, long currentTierBaseAddr) { + boolean interrupted = false; + long startPos = 0L; + CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup; + while (!hashLookup.empty(hashLookup.readEntry(s.segmentBaseAddr, startPos))) { + startPos = hashLookup.step(startPos); + } + hlp.initHashLookupPos(startPos); + do { + hlp.setHashLookupPos(hashLookup.step(hlp.hashLookupPos)); + long entry = hashLookup.readEntry(s.segmentBaseAddr, hlp.hashLookupPos); + initHashLookupEntry(entry); + if (!hashLookup.empty(entry)) { + e.readExistingEntry(hashLookup.value(entry)); + if (entryIsPresent()) { + initEntryRemovedOnThisIteration(false); + try { if (!action.test((E) this)) { interrupted = true; break; @@ -80,15 +120,16 @@ public abstract class HashSegmentIteration> if (--size == 0) break; } - + } finally { + // if doReplaceValue() -> relocation() -> alloc() -> nextTier() + // was called, restore the tier we were iterating over + if (s.segmentTier != currentTier) + s.initSegmentTier_WithBaseAddr(currentTier, currentTierBaseAddr); } } - } while (hlp.hashLookupPos != startPos); - return !interrupted; - } finally { - s.innerReadLock.unlock(); - initEntryRemovedOnThisIteration(false); - } + } + } while (hlp.hashLookupPos != startPos); + return interrupted ? ~size : size; } @Override @@ -118,7 +159,7 @@ public abstract class HashSegmentIteration> public void iterationRemove() { // this condition mean -- some other entry taken place of the removed one - if (hh.h().hashLookup.remove(s.segmentBase, hlp.hashLookupPos) != hlp.hashLookupPos) { + if (hh.h().hashLookup.remove(s.segmentBaseAddr, hlp.hashLookupPos) != hlp.hashLookupPos) { // if so, should make step back, to compensate step forward on the next iteration, // to consume the shifted entry hlp.setHashLookupPos(hh.h().hashLookup.stepBack(hlp.hashLookupPos)); diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java new file mode 100644 index 000000000..f809b9cc7 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java @@ -0,0 +1,50 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.hash.impl.stage.iter; + +import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; +import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; +import net.openhft.chronicle.hash.impl.stage.entry.HashLookupSearch; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; + +@Staged +public abstract class IterationSegmentStages extends SegmentStages { + + @StageRef VanillaChronicleHashHolder hh; + @StageRef HashSegmentIteration it; + @StageRef HashLookupSearch hls; + + /** + * During iteration, nextTier() is called in doReplaceValue() -> relocation() -> alloc(). + * When the entry is relocated to the next tier, an entry should be inserted into hash + * lookup. To insert an entry into hashLookup, should 1) locate empty slot, see {@link + * KeySearch#initKeySearch()}, and 2) know the part of the hash code to insert, we know + * it during iteration + */ + @Override + public void nextTier() { + super.nextTier(); + hls.initSearchKey(hh.h().hashLookup.key(it.hashLookupEntry)); + } + + public void initSegmentTier_WithBaseAddr(int tier, long tierBaseAddr) { + segmentTier = tier; + segmentBaseAddr = tierBaseAddr; + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java index 49746607b..be9333dd8 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java @@ -24,8 +24,10 @@ import net.openhft.chronicle.hash.impl.stage.data.bytes.InputKeyBytesData; import net.openhft.chronicle.hash.impl.stage.data.instance.InputKeyInstanceData; import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages; import net.openhft.chronicle.hash.impl.stage.entry.HashLookupPos; +import net.openhft.chronicle.hash.impl.stage.entry.HashLookupSearch; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; +import net.openhft.sg.Stage; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -72,8 +74,48 @@ public abstract class HashQuery implements HashEntry { hashOfKey = ks.inputKey.hash(LongHashFunction.city_1_1()); } + public enum EntryPresence {PRESENT, ABSENT} + + @Stage("PresenceOfEntry") private EntryPresence entryPresence = null; + + private void initPresenceOfEntry() { + if (ks.searchStatePresent() || tieredEntryPresent()) { + entryPresence = EntryPresence.PRESENT; + } else { + entryPresence = EntryPresence.ABSENT; + } + } + + public void initPresenceOfEntry(EntryPresence entryPresence) { + this.entryPresence = entryPresence; + } + + private boolean tieredEntryPresent() { + int firstTier = s.segmentTier; + long firstTierBaseAddr = s.segmentBaseAddr; + while (true) { + if (s.hasNextTier()) { + s.nextTier(); + } else { + if (s.segmentTier != 0) + s.initSegmentTier(); // loop to the root tier + } + if (s.segmentBaseAddr == firstTierBaseAddr) + break; + if (ks.searchStatePresent()) + return true; + } + // not found + if (firstTier != 0) { + // key is absent; probably are going to allocate a new entry; + // start trying from the root tier + s.initSegmentTier(); + } + return false; + } + public boolean entryPresent() { - return ks.searchStatePresent(); + return entryPresence == EntryPresence.PRESENT; } @Override @@ -86,6 +128,7 @@ public abstract class HashQuery implements HashEntry { hashLookupSearch.remove(); entry.innerRemoveEntryExceptHashLookupUpdate(); ks.setSearchState(DELETED); + initPresenceOfEntry(EntryPresence.ABSENT); } else { throw new IllegalStateException("Entry is absent when doRemove() is called"); } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/KeySearch.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/KeySearch.java index 7d1f1f88f..1829b7438 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/KeySearch.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/KeySearch.java @@ -19,6 +19,7 @@ package net.openhft.chronicle.hash.impl.stage.query; import net.openhft.chronicle.bytes.BytesUtil; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages; +import net.openhft.chronicle.hash.impl.stage.entry.HashLookupSearch; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.sg.Stage; import net.openhft.sg.StageRef; @@ -36,6 +37,8 @@ public abstract class KeySearch { public Data inputKey = null; + abstract boolean inputKeyInit(); + public void initInputKey(Data inputKey) { this.inputKey = inputKey; } @@ -55,21 +58,27 @@ public abstract class KeySearch { this.searchState = newSearchState; } - void initKeySearch() { + public void initKeySearch() { for (long pos; (pos = hashLookupSearch.nextPos()) >= 0L;) { - entry.readExistingEntry(pos); - if (!keyEquals()) - continue; - hashLookupSearch.found(); - keyFound(); - return; + // otherwise we are inside iteration relocation. + // During iteration, key search occurs when doReplaceValue() exhausts space in + // the current segment, and insertion into the tiered segment requires to locate + // an empty slot in the hashLookup. + if (inputKeyInit()) { + entry.readExistingEntry(pos); + if (!keyEquals()) + continue; + hashLookupSearch.found(); + keyFound(); + return; + } } searchState = SearchState.ABSENT; } boolean keyEquals() { return inputKey.size() == entry.keySize && - BytesUtil.bytesEqual(entry.entryBS, entry.keyOffset, + BytesUtil.bytesEqual(s.segmentBS, entry.keyOffset, inputKey.bytes(), inputKey.offset(), entry.keySize); } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupPos.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupPos.java index 8e08e3640..c10a2c8fd 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupPos.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupPos.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.hash.impl.stage.query; import net.openhft.chronicle.hash.impl.stage.entry.HashLookupPos; +import net.openhft.chronicle.hash.impl.stage.entry.HashLookupSearch; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -28,6 +29,10 @@ public abstract class QueryHashLookupPos extends HashLookupPos { @StageRef HashLookupSearch hashLookupSearch; public void initHashLookupPos() { + // validation + make hashLookupPos a dependant of segmentTier. This is needed, because + // after tier change should re-perform hashLookupSearch, starting from the searchStartPos + int segmentTier = s.segmentTier; + assert segmentTier >= 0 : "segment tier " + segmentTier; s.innerReadLock.lock(); this.hashLookupPos = hashLookupSearch.searchStartPos; } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupSearch.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupSearch.java new file mode 100644 index 000000000..907698567 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupSearch.java @@ -0,0 +1,31 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.hash.impl.stage.query; + +import net.openhft.chronicle.hash.impl.stage.entry.HashLookupSearch; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; + +@Staged +public abstract class QueryHashLookupSearch extends HashLookupSearch { + + @StageRef HashQuery op; + + void initSearchKey() { + initSearchKey(hl().maskUnsetKey(hh.h().hashSplitting.segmentHash(op.hashOfKey))); + } +} 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 be6946ee5..d15b47938 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 @@ -35,11 +35,17 @@ public class SearchAllocatedChunks extends AllocatedChunks { } } - public void initEntryAndKey(long entrySize) { + /** + * @return {@code true} is 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(); - entry.writeNewEntry(s.alloc(allocatedChunks), ks.inputKey); + int tierBeforeAllocation = s.segmentTier; + long pos = s.alloc(allocatedChunks); + entry.writeNewEntry(pos, ks.inputKey); + return s.segmentTier > tierBeforeAllocation; } } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 17158bb00..027325d40 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -49,7 +49,6 @@ import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.net.InetSocketAddress; import java.util.ArrayList; -import java.util.Arrays; import java.util.List; import java.util.Objects; import java.util.concurrent.ExecutorService; @@ -158,6 +157,8 @@ public final class ChronicleMapBuilder implements private long lockTimeOut = 20000L; private TimeUnit lockTimeOutUnit = TimeUnit.MILLISECONDS; private int metaDataBytes = 0; + private double maxBloatFactor = 1.0; + private boolean putReturnsNull = false; private boolean removeReturnsNull = false; @@ -915,10 +916,12 @@ public final class ChronicleMapBuilder implements // // The only problem with small segments is that due to probability theory, if there are // a lot of segments each of little number of entries, difference between most filled - // and least filled segment in the Chronicle Map grows. It is critical, because we don't - // support segment resize yet, and are required to allocate unnecessarily many entries per - // each segment. To compensate this at least on linux, don't accept segment sizes that - // with the given entry sizes, lead to too small total segment sizes in native memory pages, + // and least filled segment in the Chronicle Map grows. (Number of entries in a segment is + // Poisson-distributed with mean = average number of entries per segment.) It is meaningful, + // because segment tiering is exceptional mechanism, only very few segments should be + // tiered, if any, normally. So, we are required to allocate unnecessarily many entries per + // each segment. To compensate this at least on linux, don't accept segment sizes that with + // the given entry sizes, lead to too small total segment sizes in native memory pages, // see comment in tryHashLookupSlotSize() for (int hashLookupSlotSize = 3; hashLookupSlotSize <= 7; hashLookupSlotSize++) { long segments = tryHashLookupSlotSize(hashLookupSlotSize, replicated); @@ -1020,6 +1023,25 @@ public final class ChronicleMapBuilder implements return metaDataBytes; } + @Override + public ChronicleMapBuilder maxBloatFactor(double maxBloatFactor) { + if (Double.isNaN(maxBloatFactor) || maxBloatFactor < 1.0 || maxBloatFactor > 1_000.0) { + throw new IllegalArgumentException("maxBloatFactor should be in [1.0, 1_000.0] " + + "bounds, " + maxBloatFactor + " given"); + } + this.maxBloatFactor = maxBloatFactor; + return this; + } + + long maxExtraTiers(boolean replicated) { + int actualSegments = actualSegments(replicated); + // maxBloatFactor is scale, so we do (- 1.0) to compute _extra_ tiers + return ((long) (maxBloatFactor - 1.0) * actualSegments) + // but to mitigate slight misconfiguration, and uneven distribution of entries + // between segments, add 1.0 x actualSegments + + actualSegments; + } + @Override public String toString() { return "ChronicleMapBuilder{" + @@ -1414,7 +1436,7 @@ public final class ChronicleMapBuilder implements (VanillaChronicleMap) m; map.initTransientsFromBuilder(this); map.headerSize = roundUpMapHeaderSize(fis.getChannel().position()); - long sizeInBytes = map.sizeInBytes(); + long sizeInBytes = map.sizeInBytesWithoutTiers(); if (sizeInBytes != fileLength) { throw new IOException("The file " + file + "the map is serialized from " + "has unexpected length " + fileLength + ", probably corrupted. " + @@ -1501,11 +1523,11 @@ public final class ChronicleMapBuilder implements // pushingToMapEventListener(); VanillaChronicleMap map = newMap(singleHashReplication, channel); // TODO this method had been moved -// if(OS.warnOnWindows(map.sizeInBytes())){ +// if(OS.warnOnWindows(map.sizeInBytesWithoutTiers())){ // throw new IllegalStateException("Windows cannot support this configuration"); // } BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE, - map.sizeInBytes(), true); + map.sizeInBytesWithoutTiers(), true); map.createMappedStoreAndSegments(bytesStore); return establishReplication(map, singleHashReplication, channel); } catch (IOException e) { diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderPrivateAPI.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderPrivateAPI.java index ac9f82c1a..ff860f292 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderPrivateAPI.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderPrivateAPI.java @@ -66,4 +66,9 @@ class ChronicleMapBuilderPrivateAPI implements ChronicleHashBuilderPrivateAPI public int actualSegments(boolean replicated) { return b.actualSegments(replicated); } + + @Override + public long maxExtraTiers(boolean replicated) { + return b.maxExtraTiers(replicated); + } } diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index a0956893f..98379edeb 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -16,18 +16,21 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.ReplicatedHashSegmentContext; +import net.openhft.chronicle.algo.bitset.BitSetFrame; +import net.openhft.chronicle.algo.bitset.ConcurrentFlatBitSetFrame; +import net.openhft.chronicle.algo.bitset.SingleThreadedFlatBitSetFrame; +import net.openhft.chronicle.hash.impl.TierData; +import net.openhft.chronicle.hash.impl.VanillaChronicleHash; import net.openhft.chronicle.hash.replication.AbstractReplication; import net.openhft.chronicle.hash.replication.ReplicableEntry; import net.openhft.chronicle.hash.replication.TimeProvider; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.chronicle.map.impl.CompiledReplicatedMapIterationContext; import net.openhft.chronicle.map.impl.CompiledReplicatedMapQueryContext; +import net.openhft.chronicle.map.impl.IterationContextInterface; import net.openhft.chronicle.map.replication.MapRemoteOperations; -import net.openhft.chronicle.map.replication.MapReplicableEntry; import net.openhft.lang.Maths; import net.openhft.lang.collection.ATSDirectBitSet; -import net.openhft.lang.collection.SingleThreadedDirectBitSet; import net.openhft.lang.io.Bytes; import net.openhft.lang.model.DataValueClasses; import org.jetbrains.annotations.NotNull; @@ -39,10 +42,10 @@ import java.io.IOException; import java.io.ObjectInputStream; import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReferenceArray; +import static net.openhft.chronicle.algo.bytes.Access.nativeAccess; import static net.openhft.lang.MemoryUnit.*; import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND; @@ -115,6 +118,11 @@ public class ReplicatedChronicleMap remoteOpContext; transient CompiledReplicatedMapIterationContext remoteItContext; + transient BitSetFrame mainSegmentsModIterFrameForUpdates; + transient BitSetFrame mainSegmentsModIterFrameForIteration; + transient BitSetFrame tierBulkModIterFrameForUpdates; + transient BitSetFrame tierBulkModIterFrameForIteration; + public ReplicatedChronicleMap(@NotNull ChronicleMapBuilder builder, AbstractReplication replication) throws IOException { @@ -170,6 +178,16 @@ public class ReplicatedChronicleMap(127 + RESERVED_MOD_ITER); closeables = new CopyOnWriteArraySet<>(); + long mainSegmentsBitSetSize = BYTES.toBits(modIterBitSetSizeInBytes()); + mainSegmentsModIterFrameForUpdates = + new SingleThreadedFlatBitSetFrame(mainSegmentsBitSetSize); + mainSegmentsModIterFrameForIteration = + new ConcurrentFlatBitSetFrame(mainSegmentsBitSetSize); + + long tierBulkBitSetSize = + BYTES.toBits(tierBulkModIterBitSetSizeInBytes(numberOfTiersInBulk)); + tierBulkModIterFrameForUpdates = new SingleThreadedFlatBitSetFrame(tierBulkBitSetSize); + tierBulkModIterFrameForIteration = new ConcurrentFlatBitSetFrame(tierBulkBitSetSize); } private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { @@ -182,6 +200,16 @@ public class ReplicatedChronicleMap c.context().remove(c)); - } - void addCloseable(Closeable closeable) { closeables.add(closeable); } @@ -284,61 +307,56 @@ public class ReplicatedChronicleMap 0L; next = modIterSet.nextSetBit(next + 1L)) { try { - acquireModificationIterator((byte) next).raiseChange(segmentIndex, pos, timestamp); + acquireModificationIterator((byte) next).raiseChange(tierIndex, pos, timestamp); } catch (Exception e) { LOG.error("", e); } } } - public void dropChange(long segmentIndex, long pos) { + public void dropChange(long tierIndex, long pos) { for (long next = modIterSet.nextSetBit(0L); next > 0L; next = modIterSet.nextSetBit(next + 1L)) { try { - acquireModificationIterator((byte) next).dropChange(segmentIndex, pos); + acquireModificationIterator((byte) next).dropChange(tierIndex, pos); } catch (Exception e) { LOG.error("", e); } } } - public void moveChange(long segmentIndex, long oldPos, long newPos, long timestamp) { + public void moveChange(long oldTierIndex, long oldPos, long newTierIndex, long newPos, + long timestamp) { for (long next = modIterSet.nextSetBit(0L); next > 0L; next = modIterSet.nextSetBit(next + 1L)) { try { ModificationIterator modificationIterator = acquireModificationIterator((byte) next); - if (modificationIterator.dropChange(segmentIndex, oldPos)) - modificationIterator.raiseChange(segmentIndex, newPos, timestamp); + if (modificationIterator.dropChange(oldTierIndex, oldPos)) + modificationIterator.raiseChange(newTierIndex, newPos, timestamp); } catch (Exception e) { LOG.error("", e); } } } - public boolean isChanged(long segmentIndex, long pos) { + public boolean isChanged(long tierIndex, long pos) { for (long next = modIterSet.nextSetBit(0L); next > 0L; next = modIterSet.nextSetBit(next + 1L)) { ModificationIterator modificationIterator = acquireModificationIterator((byte) next); - if (modificationIterator.isChanged(segmentIndex, pos)) + if (modificationIterator.isChanged(tierIndex, pos)) return true; } return false; @@ -527,14 +545,13 @@ public class ReplicatedChronicleMap timeProvider.currentTime() - TimeUnit.SECONDS.toMillis(1) && // timestamp <= timeProvider.currentTime() : "timeStamp=" + timestamp + ", " + // "currentTime=" + timeProvider.currentTime(); @@ -586,13 +626,53 @@ public class ReplicatedChronicleMap> log2NumberOfTiersInBulk; + TierBulkData tierBulkData = tierBulkOffsets.get((int) bulkIndex); + long bitSetAddr = tierBulkData.langBytes.address() + tierBulkData.offset + + offsetToBitSetWithinATierBulk; + tierBulkModIterFrameForUpdates.set(nativeAccess(), null, bitSetAddr, bitIndex); } - boolean isChanged(long segmentIndex, long pos) { - return changesForUpdates.isSet(combine(segmentIndex, pos)); + boolean dropChange(long tierIndex, long pos) { + LOG.debug("drop change: id {}, tierIndex {}, pos {}", localIdentifier, tierIndex, pos); + long bitIndex = combine(tierIndex, pos); + if (tierIndex <= actualSegments) { + return mainSegmentsModIterFrameForUpdates.clearIfSet(nativeAccess(), null, + mainSegmentsChangesBitSetAddr, bitIndex); + } else { + return dropExtraTierChange(tierIndex, bitIndex); + } + } + + private boolean dropExtraTierChange(long tierIndex, long bitIndex) { + tierIndex = tierIndex - actualSegments - 1; + long bulkIndex = tierIndex >> log2NumberOfTiersInBulk; + TierBulkData tierBulkData = tierBulkOffsets.get((int) bulkIndex); + long bitSetAddr = tierBulkData.langBytes.address() + tierBulkData.offset + + offsetToBitSetWithinATierBulk; + return tierBulkModIterFrameForUpdates.clearIfSet(nativeAccess(), null, + bitSetAddr, bitIndex); + } + + boolean isChanged(long tierIndex, long pos) { + long bitIndex = combine(tierIndex, pos); + if (tierIndex <= actualSegments) { + return mainSegmentsModIterFrameForUpdates.isSet(nativeAccess(), null, + mainSegmentsChangesBitSetAddr, bitIndex); + } else { + return isChangedExtraTier(tierIndex, bitIndex); + } + } + + private boolean isChangedExtraTier(long tierIndex, long bitIndex) { + tierIndex = tierIndex - actualSegments - 1; + long bulkIndex = tierIndex >> log2NumberOfTiersInBulk; + TierBulkData tierBulkData = tierBulkOffsets.get((int) bulkIndex); + long bitSetAddr = tierBulkData.langBytes.address() + tierBulkData.offset + + offsetToBitSetWithinATierBulk; + return tierBulkModIterFrameForUpdates.isSet(nativeAccess(), null, bitSetAddr, bitIndex); } /** @@ -604,10 +684,47 @@ public class ReplicatedChronicleMap 0L && changesForIteration.nextSetBit(0L) != NOT_FOUND); + return nextPosition() >= 0; + } + + private long nextPosition() { + long nextPos; + long position = this.position; + boolean allBitSetsScannedFromTheStart = false; + // at most 2 iterations + while (!allBitSetsScannedFromTheStart) { + if (iterationMainSegmentsAreaOrTierBulk == -1) { + allBitSetsScannedFromTheStart = position < 0; + if ((nextPos = mainSegmentsModIterFrameForIteration.nextSetBit(nativeAccess(), + null, mainSegmentsChangesBitSetAddr, position + 1)) != NOT_FOUND) { + return nextPos; + } else { + this.position = position = NOT_FOUND; + // go to the first bulk + iterationMainSegmentsAreaOrTierBulk = 0; + } + } + // for each allocated tier bulk + while (iterationMainSegmentsAreaOrTierBulk < extraAllocatedTierBulkCount()) { + if (true) + throw new AssertionError("shouldn't be here yet"); + VanillaChronicleHash.TierBulkData tierBulkData = + tierBulkOffsets.get(iterationMainSegmentsAreaOrTierBulk); + tierBulkBitSetAddr = tierBulkData.langBytes.address() + tierBulkData.offset + + offsetToBitSetWithinATierBulk; + if ((nextPos = tierBulkModIterFrameForIteration.nextSetBit(nativeAccess(), null, + tierBulkBitSetAddr, position + 1)) != NOT_FOUND) { + return nextPos; + } + // go to the next bulk + iterationMainSegmentsAreaOrTierBulk++; + this.position = position = NOT_FOUND; + } + this.iterationMainSegmentsAreaOrTierBulk = -1; + this.position = position = NOT_FOUND; + bootStrapTimeStamp.set(0); + } + return NOT_FOUND; } /** @@ -616,28 +733,36 @@ public class ReplicatedChronicleMap>> segmentIndexShift); + int tierIndexMinusOne = (int) (position >>> segmentIndexShift); + try (CompiledReplicatedMapIterationContext context = iterationContext()) { - context.initSegmentIndex(segmentIndex); + if (iterationMainSegmentsAreaOrTierBulk < 0) { + // if main area + context.initSegmentIndex(tierIndexMinusOne); + } else { + long tierBaseAddr = tierIndexToBaseAddr(tierIndexMinusOne + 1); + long tierDataAddr = tierBaseAddr + segmentHashLookupOuterSize; + context.initSegmentIndex(TierData.segmentIndex(tierDataAddr)); + int tier = TierData.tier(tierDataAddr); + long tierIndex = actualSegments + + (iterationMainSegmentsAreaOrTierBulk << log2NumberOfTiersInBulk) + + tierIndexMinusOne + 1; + context.initSegmentTier(tier, tierIndex, tierBaseAddr); + } + context.updateLock().lock(); - if (changesForUpdates.get(position)) { + + if (changesForUpdatesGet(position)) { entryCallback.onBeforeEntry(); @@ -646,20 +771,20 @@ public class ReplicatedChronicleMap { if (debugEnabled) { @@ -708,7 +852,7 @@ public class ReplicatedChronicleMap= fromTimeStamp && (!bootstrapOnlyLocalEntries || e.originIdentifier() == localIdentifier)) { - raiseChange(segmentIndex, c.pos(), c.timestamp()); + raiseChange(c.tierIndex(), c.pos(), c.timestamp()); } }); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java index 71261b8bf..86786574d 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java @@ -21,6 +21,8 @@ import net.openhft.chronicle.hash.impl.stage.entry.*; import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop; import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder; import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder; +import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceData; import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages; @@ -43,10 +45,12 @@ import net.openhft.sg.Staged; MapSegmentIteration.class, KeyBytesInterop.class, - SegmentStages.class, + IterationSegmentStages.class, HashLookupPos.class, IterationCheckOnEachPublicOperation.class, AllocatedChunks.class, + KeySearch.class, + HashLookupSearch.class, WrappedValueInstanceValueHolder.class, MapEntryStages.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java index 10ef73341..9afefde58 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java @@ -51,7 +51,7 @@ import net.openhft.sg.Staged; KeyBytesInterop.class, QuerySegmentStages.class, KeySearch.class, - HashLookupSearch.class, + QueryHashLookupSearch.class, QueryHashLookupPos.class, QueryCheckOnEachPublicOperation.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/QueryContextInterface.java b/src/main/java/net/openhft/chronicle/map/impl/QueryContextInterface.java index bdf13f07c..691944312 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/QueryContextInterface.java +++ b/src/main/java/net/openhft/chronicle/map/impl/QueryContextInterface.java @@ -41,6 +41,4 @@ public interface QueryContextInterface extends ExternalMapQueryContext< void initSegmentIndex(int segmentIndex); boolean segmentIndexInit(); - - void clear(); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java index 9dc0e2674..69ce8b564 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java @@ -21,6 +21,8 @@ import net.openhft.chronicle.hash.impl.stage.entry.*; import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop; import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder; import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder; +import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceData; @@ -30,7 +32,6 @@ import net.openhft.chronicle.map.impl.stage.iter.ReplicatedMapAbsentDelegatingFo import net.openhft.chronicle.map.impl.stage.iter.ReplicatedMapEntryDelegating; import net.openhft.chronicle.map.impl.stage.iter.ReplicatedMapSegmentIteration; import net.openhft.chronicle.map.impl.stage.map.*; -import net.openhft.chronicle.map.impl.stage.query.ReplicatedMapAbsent; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; import net.openhft.sg.Context; import net.openhft.sg.Staged; @@ -45,10 +46,12 @@ import net.openhft.sg.Staged; ReplicatedMapSegmentIteration.class, KeyBytesInterop.class, - SegmentStages.class, + IterationSegmentStages.class, HashLookupPos.class, IterationCheckOnEachPublicOperation.class, AllocatedChunks.class, + KeySearch.class, + HashLookupSearch.class, WrappedValueInstanceValueHolder.class, ReplicatedMapEntryStages.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java index 00937a4dc..85e080f3c 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java @@ -27,7 +27,10 @@ import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder; import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder; import net.openhft.chronicle.hash.impl.stage.input.HashInputBytes; -import net.openhft.chronicle.hash.impl.stage.query.*; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; +import net.openhft.chronicle.hash.impl.stage.query.QueryHashLookupPos; +import net.openhft.chronicle.hash.impl.stage.query.QueryHashLookupSearch; +import net.openhft.chronicle.hash.impl.stage.query.SearchAllocatedChunks; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.data.bytes.*; import net.openhft.chronicle.map.impl.stage.data.instance.InputValueInstanceData; @@ -57,7 +60,7 @@ import net.openhft.sg.Staged; KeyBytesInterop.class, ReplicatedQuerySegmentStages.class, KeySearch.class, - HashLookupSearch.class, + QueryHashLookupSearch.class, QueryHashLookupPos.class, QueryCheckOnEachPublicOperation.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/EntryValueBytesData.java b/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/EntryValueBytesData.java index 51d40fb38..8ab2ed483 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/EntryValueBytesData.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/EntryValueBytesData.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.map.impl.stage.data.bytes; import net.openhft.chronicle.bytes.RandomDataInput; import net.openhft.chronicle.hash.AbstractData; +import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages; import net.openhft.chronicle.map.impl.stage.map.ValueBytesInterop; @@ -29,6 +30,7 @@ import net.openhft.sg.Staged; public class EntryValueBytesData extends AbstractData { @StageRef ValueBytesInterop vi; + @StageRef SegmentStages s; @StageRef MapEntryStages entry; @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation; @@ -43,7 +45,7 @@ public class EntryValueBytesData extends AbstractData { @Override public RandomDataInput bytes() { checkOnEachPublicOperation.checkOnEachPublicOperation(); - return entry.entryBS; + return s.segmentBS; } @Override @@ -71,7 +73,7 @@ public class EntryValueBytesData extends AbstractData { } private V innerGetUsing(V usingValue) { - entry.entryBytes.position(entry.valueOffset); - return vi.valueReader.read(entry.entryBytes, size(), usingValue); + s.segmentBytes.position(entry.valueOffset); + return vi.valueReader.read(s.segmentBytes, size(), usingValue); } } 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 850af9548..95fac487e 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 @@ -17,8 +17,10 @@ package net.openhft.chronicle.map.impl.stage.entry; import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; import net.openhft.chronicle.hash.impl.stage.entry.AllocatedChunks; import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.MapEntry; import net.openhft.chronicle.map.VanillaChronicleMap; import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder; @@ -34,6 +36,7 @@ public abstract class MapEntryStages extends HashEntryStages @StageRef public VanillaChronicleMapHolder mh; @StageRef public AllocatedChunks allocatedChunks; + @StageRef KeySearch ks; long countValueSizeOffset() { return keyEnd(); @@ -50,20 +53,20 @@ public abstract class MapEntryStages extends HashEntryStages @Stage("ValueSize") private void countValueOffset() { - mh.m().alignment.alignPositionAddr(entryBytes); - valueOffset = entryBytes.position(); + mh.m().alignment.alignPositionAddr(s.segmentBytes); + valueOffset = s.segmentBytes.position(); } void initValueSize(long valueSize) { this.valueSize = valueSize; - entryBytes.position(valueSizeOffset); - mh.m().valueSizeMarshaller.writeSize(entryBytes, valueSize); + s.segmentBytes.position(valueSizeOffset); + mh.m().valueSizeMarshaller.writeSize(s.segmentBytes, valueSize); countValueOffset(); } void initValueSize() { - entryBytes.position(valueSizeOffset); - valueSize = mh.m().readValueSize(entryBytes); + s.segmentBytes.position(valueSizeOffset); + valueSize = mh.m().readValueSize(s.segmentBytes); countValueOffset(); } @@ -73,13 +76,13 @@ public abstract class MapEntryStages extends HashEntryStages } public void initValue(Data value) { - entryBytes.position(valueSizeOffset); + s.segmentBytes.position(valueSizeOffset); initValueSize(value.size()); writeValue(value); } public void writeValue(Data value) { - value.writeTo(entryBS, valueOffset); + value.writeTo(s.segmentBS, valueOffset); } public void initValue_WithoutSize( @@ -138,15 +141,15 @@ public abstract class MapEntryStages extends HashEntryStages "entry takes " + newSizeInChunks + " chunks, " + m.maxChunksPerEntry + " is maximum."); } - if (s.freeList.allClear(pos + entrySizeInChunks, pos + newSizeInChunks)) { - s.freeList.set(pos + entrySizeInChunks, pos + newSizeInChunks); + if (s.freeList.isRangeClear(pos + entrySizeInChunks, pos + newSizeInChunks)) { + s.freeList.setRange(pos + entrySizeInChunks, pos + newSizeInChunks); break newValueDoesNotFit; } relocation(newValue, newSizeOfEverythingBeforeValue); return; } else if (newSizeInChunks < entrySizeInChunks) { // Freeing extra chunks - s.freeList.clear(pos + newSizeInChunks, pos + entrySizeInChunks); + s.freeList.clearRange(pos + newSizeInChunks, pos + entrySizeInChunks); // Do NOT reset nextPosToSearchFrom, because if value // once was larger it could easily became larger again, // But if these chunks will be taken by that time, @@ -188,8 +191,30 @@ public abstract class MapEntryStages extends HashEntryStages s.innerWriteLock.lock(); s.free(pos, entrySizeInChunks); long entrySize = innerEntrySize(newSizeOfEverythingBeforeValue, newValue.size()); - allocatedChunks.initEntryAndKeyCopying(entrySize, valueSizeOffset - keySizeOffset); - writeValueAndPutPos(newValue); + // need to copy, because in initEntryAndKeyCopying(), in alloc(), nextTier() called -> + // hashLookupPos cleared, as a dependant + long oldHashLookupPos = hlp.hashLookupPos; + long oldHashLookupAddr = s.segmentBaseAddr; + + boolean tierHasChanged = + allocatedChunks.initEntryAndKeyCopying(entrySize, valueSizeOffset - keySizeOffset); + + if (tierHasChanged) { + // implicitly inits key search, locating hashLookupPos on the empty slot + assert ks.searchStateAbsent(); + } + + initValue(newValue); + + freeExtraAllocatedChunks(); + + CompactOffHeapLinearHashTable hl = hh.h().hashLookup; + long oldEntry = hl.readEntry(oldHashLookupAddr, oldHashLookupPos); + hl.checkValueForPut(pos); + hl.writeEntryVolatile(s.segmentBaseAddr, hlp.hashLookupPos, + oldEntry, hl.key(oldEntry), pos); + if (tierHasChanged) + hl.remove(oldHashLookupAddr, oldHashLookupPos); } public final long entrySize(long keySize, long valueSize) { diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java index 4d0fd9e49..ed4c1b2e1 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java @@ -43,9 +43,9 @@ public abstract class ReplicatedMapEntryStages extends MapEntryStages extends MapEntryStages extends MapEntryStages extends MapEntryStages extends MapEntryStages newValue, long newSizeOfEverythingBeforeValue) { long oldPos = pos; + long oldTierIndex = s.tierIndex; super.relocation(newValue, newSizeOfEverythingBeforeValue); - ru.moveChange(oldPos, pos); + ru.moveChange(oldTierIndex, oldPos, pos); } private boolean testTimeStampInSensibleRange() { 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 cddf981c0..1b9c45ce2 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 @@ -73,13 +73,13 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI boolean interrupted = false; long startPos = 0L; CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup; - while (!hashLookup.empty(hashLookup.readEntry(s.segmentBase, startPos))) { + while (!hashLookup.empty(hashLookup.readEntry(s.segmentBaseAddr, startPos))) { startPos = hashLookup.step(startPos); } hlp.initHashLookupPos(startPos); do { hlp.setHashLookupPos(hashLookup.step(hlp.hashLookupPos)); - long entry = hashLookup.readEntry(s.segmentBase, hlp.hashLookupPos); + long entry = hashLookup.readEntry(s.segmentBaseAddr, hlp.hashLookupPos); if (!hashLookup.empty(entry)) { e.readExistingEntry(hashLookup.value(entry)); ReplicableEntry e = !this.e.entryDeleted() ? entryDelegating : diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java index 030585c6b..eece44d3a 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java @@ -19,7 +19,8 @@ package net.openhft.chronicle.map.impl.stage.query; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; -import net.openhft.chronicle.hash.impl.stage.query.HashLookupSearch; +import net.openhft.chronicle.hash.impl.stage.entry.HashLookupSearch; +import net.openhft.chronicle.hash.impl.stage.query.HashQuery.EntryPresence; import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.MapAbsentEntry; import net.openhft.chronicle.map.MapContext; @@ -76,6 +77,7 @@ public abstract class MapAbsent implements MapAbsentEntry { } s.incrementModCount(); ks.setSearchState(PRESENT); + q.initPresenceOfEntry(EntryPresence.PRESENT); } else { throw new IllegalStateException( "Entry is present in the map when doInsert() is called"); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java index f8a0722a3..ac58868a7 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java @@ -92,6 +92,7 @@ public abstract class MapQuery extends HashQuery e.innerDefaultReplaceValue(newValue); s.incrementModCount(); ks.setSearchState(PRESENT); + initPresenceOfEntry(EntryPresence.PRESENT); } else { throw new IllegalStateException( "Entry is absent in the map when doReplaceValue() is called"); 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 8985b2de6..26602d582 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 @@ -17,7 +17,7 @@ package net.openhft.chronicle.map.impl.stage.query; import net.openhft.chronicle.hash.Data; -import net.openhft.chronicle.hash.impl.stage.query.HashLookupSearch; +import net.openhft.chronicle.hash.impl.stage.entry.HashLookupSearch; import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -44,9 +44,9 @@ public abstract class QueryMapEntryStages extends MapEntryStages { } else { newSizeInChunks = entrySizeInChunks; } - if (pos + newSizeInChunks < s.freeList.size() && - s.freeList.allClear(pos, pos + newSizeInChunks)) { - s.freeList.set(pos, pos + newSizeInChunks); + if (pos + newSizeInChunks < s.freeList.logicalSize() && + s.freeList.isRangeClear(pos, pos + newSizeInChunks)) { + s.freeList.setRange(pos, pos + newSizeInChunks); s.innerWriteLock.lock(); allocatedChunks.incrementSegmentEntriesIfNeeded(); if (newValueSizeIsDifferent) { 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 229abe3f9..0686eb16e 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 @@ -17,6 +17,7 @@ package net.openhft.chronicle.map.impl.stage.query; import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.impl.stage.query.HashQuery.EntryPresence; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; import net.openhft.sg.StageRef; @@ -38,6 +39,7 @@ public abstract class ReplicatedMapAbsent extends MapAbsent { if (!ks.searchStatePresent()) { putEntry(value); ks.setSearchState(PRESENT); + q.initPresenceOfEntry(EntryPresence.PRESENT); } else { e.innerDefaultReplaceValue(value); s.deleted(s.deleted() - 1); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java index 750fa0b00..a12eb67ee 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java @@ -45,25 +45,20 @@ public abstract class ReplicatedQuerySegmentStages extends QuerySegmentStages { @Override public long alloc(int chunks) { - long ret = allocReturnCode(chunks); - if (ret >= 0) { - return ret; - } else { - ReplicatedChronicleMap map = mh.m(); - MapSegmentContext sc = map.segmentContext(segmentIndex); - ((ReplicatedHashSegmentContext) sc).forEachSegmentReplicableEntry(cleanupAction); - ret = allocReturnCode(chunks); + while (true) { + long ret = allocReturnCode(chunks); if (ret >= 0) { return ret; } else { - if (chunks == 1) { - throw new IllegalStateException( - "Segment is full, no free entries found"); + ReplicatedChronicleMap map = mh.m(); + MapSegmentContext sc = map.segmentContext(segmentIndex); + ((ReplicatedHashSegmentContext) sc) + .forEachSegmentReplicableEntry(cleanupAction); + ret = allocReturnCode(chunks); + if (ret >= 0) { + return ret; } else { - throw new IllegalStateException( - "Segment is full or has no ranges of " + chunks - + " continuous free chunks" - ); + nextTier(); } } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java index 5fda8e6df..10ee3d47d 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java @@ -45,11 +45,11 @@ public abstract class ReplicationUpdate implements RemoteOperationContext } public void dropChange() { - mh.m().dropChange(s.segmentIndex, e.pos); + mh.m().dropChange(s.tierIndex, e.pos); } - public void moveChange(long oldPos, long newPos) { - mh.m().moveChange(s.segmentIndex, oldPos, newPos, e.timestamp()); + public void moveChange(long oldTierIndex, long oldPos, long newPos) { + mh.m().moveChange(oldTierIndex, oldPos, s.tierIndex, newPos, e.timestamp()); } public void updateChange() { @@ -59,11 +59,11 @@ public abstract class ReplicationUpdate implements RemoteOperationContext } public void raiseChange() { - mh.m().raiseChange(s.segmentIndex, e.pos, e.timestamp()); + mh.m().raiseChange(s.tierIndex, e.pos, e.timestamp()); } public boolean changed() { - return mh.m().isChanged(s.segmentIndex, e.pos); + return mh.m().isChanged(s.tierIndex, e.pos); } @Override diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index ea5c35708..5d30b4d20 100755 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -165,6 +165,12 @@ public final class ChronicleSetBuilder return this; } + @Override + public ChronicleSetBuilder maxBloatFactor(double maxBloatFactor) { + chronicleMapBuilder.maxBloatFactor(maxBloatFactor); + return this; + } + @Override public String toString() { return " ChronicleSetBuilder{" + diff --git a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java index 3ed4c9fa3..bac9583c6 100644 --- a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java @@ -108,7 +108,7 @@ public class EntryCountMapTest { System.out.printf(" Score: %.2f%n", scoreCount / score); } - @Test + @Test(timeout = 1000 * 60) public void testSmall() throws IOException, ExecutionException, InterruptedException { System.out.print("testSmall seeds"); int procs = Runtime.getRuntime().availableProcessors(); diff --git a/src/test/java/net/openhft/chronicle/set/Issue24ChronicleSetTest.java b/src/test/java/net/openhft/chronicle/set/Issue24ChronicleSetTest.java index ed6e5051d..466f9e0d6 100644 --- a/src/test/java/net/openhft/chronicle/set/Issue24ChronicleSetTest.java +++ b/src/test/java/net/openhft/chronicle/set/Issue24ChronicleSetTest.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.set; import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.chronicle.hash.ChronicleHashBuilder; +import org.junit.Ignore; import org.junit.Test; import java.io.File; @@ -52,7 +53,7 @@ public class Issue24ChronicleSetTest { return init(ChronicleSetBuilder.of(entryClass), entrySize, averageKeySize, fileName); } - @Test + @Test(timeout = 1000 * 60) public void issue24ChronicleSetTest() { ChronicleSet set = initSet(String.class, 1_000_000, 30, "stringSet.dat"); ExecutorService executor = Executors.newFixedThreadPool(5);