diff --git a/src/main/java/net/openhft/chronicle/common/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/common/ChronicleHashBuilder.java index 8632672c4..3036aa52f 100644 --- a/src/main/java/net/openhft/chronicle/common/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/common/ChronicleHashBuilder.java @@ -139,7 +139,7 @@ public interface ChronicleHashBuilder *
  • The number of "entries" {@linkplain #entries(long) in the whole map} and {@linkplain - * #actualEntriesPerSegment(int) per segment} is actually the number "chunks".
    • + * #actualEntriesPerSegment(long) per segment} is actually the number "chunks". * *

    Example: if values in your {@code ChronicleMap} are adjacency lists of some social graph, * where nodes are represented as {@code long} ids, and adjacency lists are serialized in @@ -195,7 +195,7 @@ public interface ChronicleHashBuilder implements Cloneable, private static final Bytes EMPTY_BYTES = new ByteBufferBytes(ByteBuffer.allocate(0)); private static final int DEFAULT_KEY_OR_VALUE_SIZE = 120; - public static final short UDP_REPLICATION_MODIFICATION_ITERATOR_ID = 128; + private static final int MAX_SEGMENTS = (1 << 30); + private static final int MAX_SEGMENTS_TO_CHAISE_COMPACT_MULTI_MAPS = (1 << 20); + + static final short UDP_REPLICATION_MODIFICATION_ITERATOR_ID = 128; private static final Logger LOG = LoggerFactory.getLogger(ChronicleMapBuilder.class.getName()); @@ -107,21 +110,21 @@ public class ChronicleMapBuilder implements Cloneable, private int minSegments = -1; private int actualSegments = -1; // used when reading the number of entries per - private int actualEntriesPerSegment = -1; + private long actualEntriesPerSegment = -1L; private int keySize = 0; private K sampleKey; private int valueSize = 0; private V sampleValue; private int entrySize = 0; private Alignment alignment = Alignment.OF_4_BYTES; - private long entries = 1 << 20; + private long entries = 1 << 20; // 1 million by default private long lockTimeOut = 2000; private TimeUnit lockTimeOutUnit = TimeUnit.MILLISECONDS; private int metaDataBytes = 0; private ChronicleHashErrorListener errorListener = ChronicleHashErrorListeners.logging(); private boolean putReturnsNull = false; private boolean removeReturnsNull = false; - private boolean largeSegments = false; + // replication private TimeProvider timeProvider = TimeProvider.SYSTEM; private BytesMarshallerFactory bytesMarshallerFactory; @@ -166,25 +169,6 @@ public class ChronicleMapBuilder implements Cloneable, } } - - @Override - public ChronicleMapBuilder minSegments(int minSegments) { - this.minSegments = minSegments; - return this; - } - - int minSegments() { - return minSegments < 1 ? tryMinSegments(4, 65536) : minSegments; - } - - private int tryMinSegments(int min, int max) { - for (int i = min; i < max; i <<= 1) { - if (i * i * i >= entries * 5 / 4) - return i; - } - return max; - } - /** * {@inheritDoc} * @@ -377,6 +361,8 @@ public class ChronicleMapBuilder implements Cloneable, @Override public ChronicleMapBuilder entries(long entries) { + if (entries <= 0L) + throw new IllegalArgumentException("Entries should be positive, " + entries + " given"); this.entries = entries; return this; } @@ -386,43 +372,112 @@ public class ChronicleMapBuilder implements Cloneable, } @Override - public ChronicleMapBuilder actualEntriesPerSegment(int actualEntriesPerSegment) { + public ChronicleMapBuilder actualEntriesPerSegment(long actualEntriesPerSegment) { + if (actualEntriesPerSegment <= 0L) + throw new IllegalArgumentException("entries per segment should be positive, " + + actualEntriesPerSegment + " given"); + if (tooManyEntriesPerSegment(actualEntriesPerSegment)) + throw new IllegalArgumentException("max entries per segment is " + + VanillaIntIntMultiMap.MAX_CAPACITY + ", " + actualEntriesPerSegment + " given"); this.actualEntriesPerSegment = actualEntriesPerSegment; return this; } - int actualEntriesPerSegment() { - if (actualEntriesPerSegment > 0) - return actualEntriesPerSegment; - int as = actualSegments(); - long actualEntries = entries + - (as == 1 ? 1 : // The extra 1 might not be needed. - as <= 8 ? entries / 12 : // 6%, (3% was min for tests) - as <= 128 ? entries / 7 : // 14% (7% was min for tests) - entries / 5); // 20% (10% was min for tests) + private boolean tooManyEntriesPerSegment(long entriesPerSegment) { + return entriesPerSegment > VanillaIntIntMultiMap.MAX_CAPACITY; + } - // round up to the next multiple of 64. - return (int) (Math.max(1, actualEntries / as) + 63) & ~63; + long actualEntriesPerSegment() { + if (actualEntriesPerSegment > 0L) + return actualEntriesPerSegment; + int actualSegments = actualSegments(); + long actualEntries = totalEntriesIfPoorDistribution(actualSegments); + long actualEntriesPerSegment = divideUpper(actualEntries, actualSegments); + if (tooManyEntriesPerSegment(actualEntriesPerSegment)) + throw new IllegalStateException("max entries per segment is " + + VanillaIntIntMultiMap.MAX_CAPACITY + " configured entries() and " + + "actualSegments() so that there should be " + actualEntriesPerSegment + + " entries per segment"); + return actualEntriesPerSegment; + } + + private long totalEntriesIfPoorDistribution(int segments) { + if (segments == 1) + return entries; + if (segments <= 8) + return entries * 13L / 12L; // 6%, (3% was min for tests) + if (segments <= 128) + return entries * 8L / 7L; // 14% (7% was min for tests) + return entries * 6L / 5L; // 20% (10% was min for tests) + } + + @Override + public ChronicleMapBuilder minSegments(int minSegments) { + checkSegments(minSegments); + this.minSegments = minSegments; + return this; + } + + int minSegments() { + return Math.max(estimateSegments(), minSegments); + } + + /** + * Heuristic -- number of segments ~= cube root from number of entries seems optimal + */ + private int estimateSegments() { + int maxSegments = 65536; + for (int i = 4; i < maxSegments; i <<= 1) { + if (((long) i) * i * i >= entries) + return i; + } + return maxSegments; } @Override public ChronicleMapBuilder actualSegments(int actualSegments) { + checkSegments(actualSegments); this.actualSegments = actualSegments; return this; } + private static void checkSegments(int segments) { + if (segments <= 0 || segments > MAX_SEGMENTS) + throw new IllegalArgumentException("segments should be positive, " + + segments + " given"); + if (segments > MAX_SEGMENTS) + throw new IllegalArgumentException("Max segments is " + MAX_SEGMENTS + ", " + + segments + " given"); + } + int actualSegments() { if (actualSegments > 0) return actualSegments; + long shortMMapSegments = trySegments(VanillaShortShortMultiMap.MAX_CAPACITY, + MAX_SEGMENTS_TO_CHAISE_COMPACT_MULTI_MAPS); + if (shortMMapSegments > 0L) + return (int) shortMMapSegments; + long intMMapSegments = trySegments(VanillaIntIntMultiMap.MAX_CAPACITY, MAX_SEGMENTS); + if (intMMapSegments > 0L) + return (int) intMMapSegments; + throw new IllegalStateException("Max segments is " + MAX_SEGMENTS + ", configured so much" + + " entries() that builder automatically decided to use " + + (-intMMapSegments) + " segments"); + } - long entriesWithMostHeadroom = entries + entries / 5; - if (!largeSegments && entriesWithMostHeadroom > (long) minSegments() << 16) { - long segments = Maths.nextPower2(entries * 5 / 6 >> 16, 128); - if (segments < 1 << 20) - return (int) segments; - } - // try to keep it 16-bit sizes segments - return (int) Maths.nextPower2(Math.max((entries >> 30) + 1, minSegments()), 1); + private long trySegments(long maxSegmentCapacity, int maxSegments) { + long segments = divideUpper(totalEntriesIfPoorDistribution(minSegments()), + maxSegmentCapacity); + segments = Maths.nextPower2(Math.max(segments, minSegments()), 1L); + return segments <= maxSegments ? segments : -segments; + } + + private static long divideUpper(long dividend, long divisor) { + return ((dividend - 1L) / divisor) + 1L; + } + + private boolean canSupportShortShort() { + return entries > (long) minSegments() << 15; } public ChronicleMapBuilder lockTimeOut(long lockTimeOut, TimeUnit unit) { @@ -497,16 +552,6 @@ public class ChronicleMapBuilder implements Cloneable, return removeReturnsNull; } - boolean largeSegments() { - return entries > 1L << (20 + 15) || largeSegments; - } - - @Override - public ChronicleMapBuilder largeSegments(boolean largeSegments) { - this.largeSegments = largeSegments; - return this; - } - @Override public ChronicleMapBuilder metaDataBytes(int metaDataBytes) { if ((metaDataBytes & 0xFF) != metaDataBytes) @@ -538,7 +583,6 @@ public class ChronicleMapBuilder implements Cloneable, ", errorListener=" + errorListener() + ", putReturnsNull=" + putReturnsNull() + ", removeReturnsNull=" + removeReturnsNull() + - ", largeSegments=" + (largeSegments ? "true" : "not configured") + ", timeProvider=" + timeProvider() + ", bytesMarshallerFactory=" + pretty(bytesMarshallerFactory) + ", objectSerializer=" + pretty(objectSerializer) + diff --git a/src/main/java/net/openhft/chronicle/map/HashSplitting.java b/src/main/java/net/openhft/chronicle/map/HashSplitting.java new file mode 100644 index 000000000..f32590d3e --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/HashSplitting.java @@ -0,0 +1,98 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import java.io.Serializable; + +import static net.openhft.lang.Maths.isPowerOf2; + +interface HashSplitting extends Serializable { + + int segmentIndex(long hash); + long segmentHash(long hash); + + static class Splitting { + static HashSplitting forSegments(int segments) { + assert segments > 0; + if (segments == 1) + return ForSingleSegment.INSTANCE; + if (isPowerOf2(segments)) + return new ForPowerOf2Segments(segments); + return new ForNonPowerOf2Segments(segments); + } + } + + enum ForSingleSegment implements HashSplitting { + INSTANCE; + + @Override + public int segmentIndex(long hash) { + return 0; + } + + @Override + public long segmentHash(long hash) { + return hash; + } + } + + static class ForPowerOf2Segments implements HashSplitting { + private static final long serialVersionUID = 0L; + + private final int mask; + private final int bits; + + ForPowerOf2Segments(int segments) { + mask = segments - 1; + bits = Integer.numberOfTrailingZeros(segments); + } + + @Override + public int segmentIndex(long hash) { + return ((int) hash) & mask; + } + + @Override + public long segmentHash(long hash) { + return hash >>> bits; + } + } + + //TODO optimize? + static class ForNonPowerOf2Segments implements HashSplitting { + private static final long serialVersionUID = 0L; + + private static final int MASK = Integer.MAX_VALUE; + private static final int BITS = 31; + + private final int segments; + + public ForNonPowerOf2Segments(int segments) { + this.segments = segments; + } + + @Override + public int segmentIndex(long hash) { + return (((int) hash) & MASK) % segments; + } + + @Override + public long segmentHash(long hash) { + return hash >>> BITS; + } + } +} diff --git a/src/main/java/net/openhft/chronicle/map/MapEventListener.java b/src/main/java/net/openhft/chronicle/map/MapEventListener.java index 98ace6a80..c33986dbc 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEventListener.java +++ b/src/main/java/net/openhft/chronicle/map/MapEventListener.java @@ -95,7 +95,7 @@ public abstract class MapEventListener> } void onRemove(M map, Bytes entry, int metaDataBytes, K key, V value, - int pos, SharedSegment segment) { + long pos, SharedSegment segment) { onRemove(map, entry, metaDataBytes, key, value); } @@ -112,7 +112,7 @@ public abstract class MapEventListener> // do nothing } - void onRelocation(int pos, SharedSegment segment) { + void onRelocation(long pos, SharedSegment segment) { // do nothing } } diff --git a/src/main/java/net/openhft/chronicle/map/MapEventListeners.java b/src/main/java/net/openhft/chronicle/map/MapEventListeners.java index 5c1b728c1..82c1ab86a 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEventListeners.java +++ b/src/main/java/net/openhft/chronicle/map/MapEventListeners.java @@ -71,7 +71,7 @@ public final class MapEventListeners { @Override public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes, - Object key, Object value, int pos, SharedSegment segment) { + Object key, Object value, long pos, SharedSegment segment) { logOperation(map, entry, metaDataBytes, " remove "); } }; @@ -93,7 +93,7 @@ public final class MapEventListeners { @Override public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes, - Object key, Object value, int pos, SharedSegment segment) { + Object key, Object value, long pos, SharedSegment segment) { LOGGER.info("{} remove {} was {}", map.file(), key, value); } }; diff --git a/src/main/java/net/openhft/chronicle/map/MultiMap.java b/src/main/java/net/openhft/chronicle/map/MultiMap.java new file mode 100644 index 000000000..60d013d1d --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/MultiMap.java @@ -0,0 +1,77 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import net.openhft.lang.collection.DirectBitSet; + +/** + * This is only used to store keys and positions, but it could store int/int key/values for another purpose. + */ +interface MultiMap { + + /** + * @return gets all the active positions as a bitset + */ + DirectBitSet getPositions(); + + /** + * Add an entry. Allow duplicate hashes, but not key/position pairs. + * + * @param key to add + * @param value to add + */ + void put(long key, long value); + + /** + * Remove a key/value pair. + * + * @param key to remove + * @param value to remove + * @throws RuntimeException if key-value pair is absent in the multi map + */ + void remove(long key, long value); + + void replace(long key, long oldValue, long newValue); + + /** + * Used for start a search for a given key + */ + void startSearch(long key); + + long getSearchHash(); + + /** + * Used for getting the next position for a given key + * + * @return the next position for the last search or negative value + */ + long nextPos(); + + void removePrevPos(); + + void replacePrevPos(long newValue); + + void putAfterFailedSearch(long value); + + void clear(); + + void forEach(EntryConsumer action); + + static interface EntryConsumer { + void accept(long key, long value); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 24e3a811c..4fddeab91 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -139,12 +139,13 @@ class ReplicatedChronicleMap, } long modIterBitSetSizeInBytes() { - return align64(bitsPerSegmentInModIterBitSet() * segments.length / 8); + return align64(bitsPerSegmentInModIterBitSet() * segments.length / 8L); } private long bitsPerSegmentInModIterBitSet() { // min 128 * 8 to prevent false sharing on updating bits from different segments - return Maths.nextPower2((long) entriesPerSegment, 128 * 8); + // TODO this doesn't prevent false sharing. There should be GAPS between per-segment bits + return Maths.nextPower2(entriesPerSegment, 128L * 8L); } @Override @@ -244,7 +245,7 @@ class ReplicatedChronicleMap, metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); long hash = metaKeyWriter.hash(keyWriter, key); int segmentNum = getSegment(hash); - int segmentHash = segmentHash(hash); + long segmentHash = segmentHash(hash); return segment(segmentNum).put(copies, metaKeyWriter, keyWriter, key, value, segmentHash, replaceIfPresent, identifier, timeStamp); } @@ -267,7 +268,7 @@ class ReplicatedChronicleMap, metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); long hash = metaKeyWriter.hash(keyWriter, key); int segmentNum = getSegment(hash); - int segmentHash = segmentHash(hash); + long segmentHash = segmentHash(hash); return segment(segmentNum).acquire(copies, metaKeyWriter, keyWriter, key, value, segmentHash, create, create ? timeProvider.currentTimeMillis() : 0L); } @@ -326,7 +327,7 @@ class ReplicatedChronicleMap, } @Override - public boolean remove(@net.openhft.lang.model.constraints.NotNull final Object key, final Object value) { + public boolean remove(@NotNull final Object key, final Object value) { if (value == null) return false; // CHM compatibility; I would throw NPE return removeIfValueIs(key, (V) value, @@ -344,13 +345,13 @@ class ReplicatedChronicleMap, metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); long hash = metaKeyWriter.hash(keyWriter, key); int segmentNum = getSegment(hash); - int segmentHash = segmentHash(hash); + long segmentHash = segmentHash(hash); return segment(segmentNum).remove(copies, metaKeyWriter, keyWriter, key, expectedValue, segmentHash, timestamp, identifier); } @Override - V replaceIfValueIs(@net.openhft.lang.model.constraints.NotNull final K key, final V existingValue, final V newValue) { + V replaceIfValueIs(@NotNull final K key, final V existingValue, final V newValue) { checkKey(key); checkValue(newValue); ThreadLocalCopies copies = keyInteropProvider.getCopies(null); @@ -360,7 +361,7 @@ class ReplicatedChronicleMap, metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); long hash = metaKeyWriter.hash(keyWriter, key); int segmentNum = getSegment(hash); - int segmentHash = segmentHash(hash); + long segmentHash = segmentHash(hash); return segment(segmentNum).replace(copies, metaKeyWriter, keyWriter, key, existingValue, newValue, segmentHash, timeProvider.currentTimeMillis()); } @@ -392,7 +393,7 @@ class ReplicatedChronicleMap, if (!isDeleted) { valueSize = valueSizeMarshaller.readSize(entry); } else { - valueSize = 0; + valueSize = 0L; } final long valuePosition = entry.position(); @@ -476,7 +477,7 @@ class ReplicatedChronicleMap, long hash = hash(source); int segmentNum = getSegment(hash); - int segmentHash = segmentHash(hash); + long segmentHash = segmentHash(hash); boolean debugEnabled = LOG.isDebugEnabled(); @@ -568,7 +569,7 @@ class ReplicatedChronicleMap, public boolean wasRemoved(@NotNull Bytes entry) { final long start = entry.position(); try { - return entry.readBoolean(keySizeMarshaller.readSize(entry) + 9); + return entry.readBoolean(keySizeMarshaller.readSize(entry) + 9L); } finally { entry.position(start); } @@ -576,8 +577,8 @@ class ReplicatedChronicleMap, class Segment extends VanillaChronicleMap.Segment { - private volatile IntIntMultiMap hashLookupLiveAndDeleted; - private volatile IntIntMultiMap hashLookupLiveOnly; + private volatile MultiMap hashLookupLiveAndDeleted; + private volatile MultiMap hashLookupLiveOnly; Segment(NativeBytes bytes, int index) { super(bytes, index); @@ -590,7 +591,7 @@ class ReplicatedChronicleMap, hashLookupLiveOnly = createMultiMap(start); } - public IntIntMultiMap getHashLookup() { + public MultiMap getHashLookup() { return hashLookupLiveOnly; } @@ -613,7 +614,7 @@ class ReplicatedChronicleMap, * @see VanillaChronicleMap.Segment#acquire */ V acquire(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, - K key, V usingValue, int hash2, boolean create, long timestamp) { + K key, V usingValue, long hash2, boolean create, long timestamp) { if (create) writeLock(); else @@ -655,7 +656,7 @@ class ReplicatedChronicleMap, private long putEntryOnAcquire(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, K key, - long keySize, int hash2, V value, boolean usingValue, + long keySize, long hash2, V value, boolean usingValue, long timestamp) { return putEntry(copies, metaKeyWriter, keyWriter, key, keySize, hash2, value, usingValue, localIdentifier, timestamp, hashLookupLiveOnly); @@ -665,13 +666,13 @@ class ReplicatedChronicleMap, /** * called from a remote node as part of replication */ - private void remoteRemove(Bytes keyBytes, int hash2, + private void remoteRemove(Bytes keyBytes, long hash2, final long timestamp, final byte identifier) { writeLock(); try { long keySize = keyBytes.remaining(); hashLookupLiveAndDeleted.startSearch(hash2); - for (int pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0; ) { + for (long pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); if (!remoteKeyEquals(keyBytes, keySize, entry)) @@ -703,8 +704,8 @@ class ReplicatedChronicleMap, } // key is not found if (LOG.isDebugEnabled()) - LOG.debug("Segment.remoteRemove() : key=" + keyBytes.toString().trim() + " was not " + - "found"); + LOG.debug("Segment.remoteRemove() : key=" + keyBytes.toString().trim() + + " was not found"); } finally { writeUnlock(); } @@ -714,7 +715,7 @@ class ReplicatedChronicleMap, /** * called from a remote node when it wishes to propagate a remove event */ - private void remotePut(@NotNull final Bytes inBytes, int hash2, + private void remotePut(@NotNull final Bytes inBytes, long hash2, final byte identifier, final long timestamp, long valuePos, long valueLimit) { writeLock(); @@ -723,7 +724,7 @@ class ReplicatedChronicleMap, final long keySize = inBytes.remaining(); hashLookupLiveAndDeleted.startSearch(hash2); - for (int pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0; ) { + for (long pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); @@ -768,7 +769,7 @@ class ReplicatedChronicleMap, // key is not found long valueSize = valueLimit - valuePos; - int pos = alloc(inBlocks(entrySize(keySize, valueSize))); + long pos = alloc(inBlocks(entrySize(keySize, valueSize))); long offset = offsetFromPos(pos); clearMetaData(offset); NativeBytes entry = entry(offset); @@ -806,15 +807,14 @@ class ReplicatedChronicleMap, } V put(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, - K key, V value, int hash2, boolean replaceIfPresent, + K key, V value, long hash2, boolean replaceIfPresent, final byte identifier, final long timestamp) { writeLock(); try { - IntIntMultiMap hashLookup = hashLookupLiveAndDeleted; + MultiMap hashLookup = hashLookupLiveAndDeleted; long keySize = metaKeyWriter.size(keyWriter, key); hashLookup.startSearch(hash2); - int pos; - while ((pos = hashLookup.nextPos()) >= 0) { + for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) @@ -878,7 +878,8 @@ class ReplicatedChronicleMap, * @param identifier the unique identifier relating to this map * @return true if the entry should not be processed */ - private boolean shouldIgnore(@NotNull final NativeBytes entry, final long timestamp, final byte identifier) { + private boolean shouldIgnore(@NotNull final NativeBytes entry, final long timestamp, + final byte identifier) { final long lastModifiedTimeStamp = entry.readLong(); @@ -916,9 +917,9 @@ class ReplicatedChronicleMap, * @see VanillaChronicleMap.Segment#putEntry */ private long putEntry(ThreadLocalCopies copies, MKI metaKeyWriter, - KI keyWriter, K key, long keySize, int hash2, V value, + KI keyWriter, K key, long keySize, long hash2, V value, boolean usingValue, final int identifier, final long timestamp, - IntIntMultiMap searchedHashLookup) { + MultiMap searchedHashLookup) { // "if-else polymorphism" is not very beautiful, but allows to // reuse the rest code of this method and doesn't hurt performance. @@ -940,7 +941,7 @@ class ReplicatedChronicleMap, } long entrySize = entrySize(keySize, valueSize); - int pos = alloc(inBlocks(entrySize)); + long pos = alloc(inBlocks(entrySize)); long offset = offsetFromPos(pos); clearMetaData(offset); NativeBytes entry = entry(offset); @@ -971,15 +972,15 @@ class ReplicatedChronicleMap, * @see VanillaChronicleMap.Segment#remove */ V remove(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, - K key, V expectedValue, int hash2, + K key, V expectedValue, long hash2, final long timestamp, final byte identifier) { assert identifier > 0; writeLock(); try { long keySize = metaKeyWriter.size(keyWriter, key); - final IntIntMultiMap multiMap = hashLookupLiveAndDeleted; + final MultiMap multiMap = hashLookupLiveAndDeleted; multiMap.startSearch(hash2); - for (int pos; (pos = multiMap.nextPos()) >= 0; ) { + for (long pos; (pos = multiMap.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) @@ -1033,7 +1034,7 @@ class ReplicatedChronicleMap, } @Override - IntIntMultiMap containsKeyHashLookup() { + MultiMap containsKeyHashLookup() { return hashLookupLiveOnly; } @@ -1041,12 +1042,12 @@ class ReplicatedChronicleMap, * @see VanillaChronicleMap.Segment#replace */ V replace(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, - K key, V expectedValue, V newValue, int hash2, long timestamp) { + K key, V expectedValue, V newValue, long hash2, long timestamp) { writeLock(); try { long keySize = metaKeyWriter.size(keyWriter, key); hashLookupLiveOnly.startSearch(hash2); - for (int pos; (pos = hashLookupLiveOnly.nextPos()) >= 0; ) { + for (long pos; (pos = hashLookupLiveOnly.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) @@ -1070,10 +1071,11 @@ class ReplicatedChronicleMap, } @Override - void replacePosInHashLookupOnRelocation(IntIntMultiMap searchedHashLookup, int prevPos, int pos) { + void replacePosInHashLookupOnRelocation(MultiMap searchedHashLookup, + long prevPos, long pos) { searchedHashLookup.replacePrevPos(pos); - int hash = searchedHashLookup.getSearchHash(); - IntIntMultiMap anotherLookup = searchedHashLookup == hashLookupLiveAndDeleted ? + long hash = searchedHashLookup.getSearchHash(); + MultiMap anotherLookup = searchedHashLookup == hashLookupLiveAndDeleted ? hashLookupLiveOnly : hashLookupLiveAndDeleted; anotherLookup.replace(hash, prevPos, pos); } @@ -1084,10 +1086,10 @@ class ReplicatedChronicleMap, this.readLock(); try { final int index = Segment.this.getIndex(); - hashLookupLiveAndDeleted.forEach(new IntIntMultiMap.EntryConsumer() { + hashLookupLiveAndDeleted.forEach(new MultiMap.EntryConsumer() { @Override - public void accept(int hash, int pos) { + public void accept(long hash, long pos) { final NativeBytes entry = entry(offsetFromPos(pos)); long keySize = keySizeMarshaller.readSize(entry); entry.skip(keySize); @@ -1154,7 +1156,7 @@ class ReplicatedChronicleMap, class EntryIterator extends VanillaChronicleMap.EntryIterator { @Override - void removePresent(VanillaChronicleMap.Segment seg, int pos) { + void removePresent(VanillaChronicleMap.Segment seg, long pos) { @SuppressWarnings("unchecked") Segment segment = (Segment) seg; @@ -1189,7 +1191,7 @@ class ReplicatedChronicleMap, return; } } - final int segmentHash = segmentHash(hash(entry, keyPosition, keyPosition + keySize)); + final long segmentHash = segmentHash(hash(entry, keyPosition, keyPosition + keySize)); segment.hashLookupLiveOnly.remove(segmentHash, pos); segment.decrementSize(); @@ -1285,7 +1287,7 @@ class ReplicatedChronicleMap, @Override public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes, - K key, V value, int pos, SharedSegment segment) { + K key, V value, long pos, SharedSegment segment) { assert ReplicatedChronicleMap.this == map : "ModificationIterator.onRemove() is called from outside of the parent map"; try { @@ -1306,7 +1308,7 @@ class ReplicatedChronicleMap, } @Override - void onRelocation(int pos, SharedSegment segment) { + void onRelocation(long pos, SharedSegment segment) { for (long next = modIterSet.nextSetBit(0); next > 0; next = modIterSet.nextSetBit(next + 1)) { try { final ModificationIterator modificationIterator = modificationIterators.get((int) next); @@ -1371,7 +1373,7 @@ class ReplicatedChronicleMap, private final EntryModifiableCallback entryModifiableCallback = new EntryModifiableCallback(); // records the current position of the cursor in the bitset - private long position = -1; + private long position = -1L; /** * @param bytes the back the bitset, used to mark which entries have changed @@ -1383,7 +1385,7 @@ class ReplicatedChronicleMap, this.modificationNotifier = modificationNotifier; long bitsPerSegment = bitsPerSegmentInModIterBitSet(); segmentIndexShift = Long.numberOfTrailingZeros(bitsPerSegment); - posMask = bitsPerSegment - 1; + posMask = bitsPerSegment - 1L; changes = new ATSDirectBitSet(bytes); } @@ -1413,7 +1415,7 @@ class ReplicatedChronicleMap, @Override public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes, - K key, V value, int pos, SharedSegment segment) { + K key, V value, long pos, SharedSegment segment) { assert ReplicatedChronicleMap.this == map : "ModificationIterator.onRemove() is called from outside of the parent map"; @@ -1427,7 +1429,7 @@ class ReplicatedChronicleMap, * Ensures that garbage in the old entry's location won't be broadcast as changed entry. */ @Override - void onRelocation(int pos, SharedSegment segment) { + void onRelocation(long pos, SharedSegment segment) { changes.clear(combine(segment.getIndex(), pos)); // don't call nextListener.onRelocation(), // because no one event listener else overrides this method. @@ -1443,8 +1445,8 @@ class ReplicatedChronicleMap, */ public boolean hasNext() { final long position = this.position; - return changes.nextSetBit(position == NOT_FOUND ? 0 : position) != NOT_FOUND || - (position > 0 && changes.nextSetBit(0) != NOT_FOUND); + return changes.nextSetBit(position == NOT_FOUND ? 0L : position) != NOT_FOUND || + (position > 0L && changes.nextSetBit(0L) != NOT_FOUND); } @@ -1456,7 +1458,7 @@ class ReplicatedChronicleMap, long position = this.position; while (true) { long oldPosition = position; - position = changes.nextSetBit(oldPosition + 1); + position = changes.nextSetBit(oldPosition + 1L); if (position == NOT_FOUND) { if (oldPosition == NOT_FOUND) { @@ -1532,7 +1534,7 @@ class ReplicatedChronicleMap, * @param segmentIndex the segment relating to the bit to set * @param pos the position relating to the bit to set */ - public void set(int segmentIndex, int pos) { + public void set(int segmentIndex, long pos) { final long combine = combine(segmentIndex, pos); changes.set(combine); } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 60db05846..13032fbc9 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -49,7 +49,7 @@ import static java.lang.Thread.currentThread; class VanillaChronicleMap, V, VW, MVW extends MetaBytesWriter> extends AbstractMap implements ChronicleMap, Serializable { - private static final long serialVersionUID = 1L; + private static final long serialVersionUID = 2L; private static final Logger LOG = LoggerFactory.getLogger(VanillaChronicleMap.class); /** @@ -80,10 +80,10 @@ class VanillaChronicleMap, final int metaDataBytes; // private final int replicas; - final int entrySize; + final long entrySize; final Alignment alignment; final int actualSegments; - final int entriesPerSegment; + final long entriesPerSegment; final MapEventListener> eventListener; // if set the ReturnsNull fields will cause some functions to return NULL // rather than as returning the Object can be expensive for something you probably don't use. @@ -98,8 +98,7 @@ class VanillaChronicleMap, transient long headerSize; transient Set> entrySet; - private int bits; - private int mask; + private final HashSplitting hashSplitting; public VanillaChronicleMap(ChronicleMapBuilder builder) throws IOException { @@ -140,29 +139,28 @@ class VanillaChronicleMap, this.metaDataBytes = builder.metaDataBytes(); this.eventListener = builder.eventListener(); - this.mask = useSmallMultiMaps() ? 0xFFFF : ~0; - this.bits = Maths.intLog2(actualSegments); + hashSplitting = HashSplitting.Splitting.forSegments(actualSegments); initTransients(); } - int segmentHash(long hash) { - return (int) (hash >>> bits) & mask; + long segmentHash(long hash) { + return hashSplitting.segmentHash(hash); } int getSegment(long hash) { - return (int) (hash & (actualSegments - 1)); + return hashSplitting.segmentIndex(hash); } /** * Cache line alignment, assuming 64-byte cache lines. */ static long align64(long l) { - return (l + 63) & ~63; + return (l + 63L) & ~63L; } - static int align8(int n) { - return (n + 7) & ~7; + static long align8(long n) { + return (n + 7L) & ~7L; } void initTransients() { @@ -252,11 +250,11 @@ class VanillaChronicleMap, } boolean useSmallMultiMaps() { - return entriesPerSegment <= (1 << 16); + return entriesPerSegment <= VanillaShortShortMultiMap.MAX_CAPACITY; } long sizeOfBitSets() { - return align64(entriesPerSegment / 8); + return align64(entriesPerSegment / 8L); } int numberOfBitSets() { @@ -270,7 +268,7 @@ class VanillaChronicleMap, + align64(sizeOfMultiMap() + sizeOfMultiMapBitSet()) * multiMapsPerSegment() + numberOfBitSets() * sizeOfBitSets() // the free list and 0+ dirty lists. + sizeOfEntriesInSegment(); - if ((ss & 63) != 0) + if ((ss & 63L) != 0) throw new AssertionError(); // Say, there is 32 KB L1 cache with 2(4, 8) way set associativity, 64-byte lines. @@ -296,7 +294,7 @@ class VanillaChronicleMap, } private long sizeOfEntriesInSegment() { - return align64((long) entriesPerSegment * entrySize); + return align64(entriesPerSegment * entrySize); } @Override @@ -347,7 +345,7 @@ class VanillaChronicleMap, metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); long hash = metaKeyInterop.hash(keyInterop, key); int segmentNum = getSegment(hash); - int segmentHash = segmentHash(hash); + long segmentHash = segmentHash(hash); return segments[segmentNum].put(copies, metaKeyInterop, keyInterop, key, value, segmentHash, replaceIfPresent); } @@ -376,7 +374,7 @@ class VanillaChronicleMap, metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); long hash = metaKeyInterop.hash(keyInterop, key); int segmentNum = getSegment(hash); - int segmentHash = segmentHash(hash); + long segmentHash = segmentHash(hash); return segments[segmentNum].acquire(copies, metaKeyInterop, keyInterop, key, value, segmentHash, create); } @@ -392,7 +390,7 @@ class VanillaChronicleMap, metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); long hash = metaKeyInterop.hash(keyInterop, key); int segmentNum = getSegment(hash); - int segmentHash = segmentHash(hash); + long segmentHash = segmentHash(hash); return segments[segmentNum].containsKey(keyInterop, metaKeyInterop, key, segmentHash); } @@ -447,7 +445,7 @@ class VanillaChronicleMap, metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); long hash = metaKeyInterop.hash(keyInterop, key); int segmentNum = getSegment(hash); - int segmentHash = segmentHash(hash); + long segmentHash = segmentHash(hash); return segments[segmentNum].remove(copies, metaKeyInterop, keyInterop, key, expectedValue, segmentHash); } @@ -472,7 +470,7 @@ class VanillaChronicleMap, @Override public long longSize() { - long result = 0; + long result = 0L; for (final Segment segment : this.segments) { result += segment.getSize(); @@ -496,8 +494,7 @@ class VanillaChronicleMap, * @param newValue the new value you wish to store in the map * @return the value that was replaced */ - V replaceIfValueIs(@net.openhft.lang.model.constraints.NotNull final K key, final V existingValue, - final V newValue) { + V replaceIfValueIs(@NotNull final K key, final V existingValue, final V newValue) { checkKey(key); checkValue(newValue); ThreadLocalCopies copies = keyInteropProvider.getCopies(null); @@ -507,7 +504,7 @@ class VanillaChronicleMap, metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); long hash = metaKeyInterop.hash(keyInterop, key); int segmentNum = getSegment(hash); - int segmentHash = segmentHash(hash); + long segmentHash = segmentHash(hash); return segments[segmentNum].replace(copies, metaKeyInterop, keyInterop, key, existingValue, newValue, segmentHash); } @@ -537,22 +534,19 @@ class VanillaChronicleMap, - encoded length of the value - bytes for the value. */ - static final int LOCK_OFFSET = 0; // 64-bit - static final int SIZE_OFFSET = LOCK_OFFSET + 8; // 32-bit - static final int PAD1_OFFSET = SIZE_OFFSET + 4; // 32-bit - static final int REPLICA_OFFSET = PAD1_OFFSET + 4; // 64-bit + static final long LOCK_OFFSET = 0L; // 64-bit + static final long SIZE_OFFSET = LOCK_OFFSET + 8L; // 32-bit final NativeBytes bytes; final MultiStoreBytes tmpBytes = new MultiStoreBytes(); final long entriesOffset; private final int index; private final SingleThreadedDirectBitSet freeList; - private IntIntMultiMap hashLookup; - private int nextPosToSearchFrom = 0; + private MultiMap hashLookup; + private long nextPosToSearchFrom = 0L; /** - * @param bytes * @param index the index of this segment held by the map */ Segment(NativeBytes bytes, int index) { @@ -563,7 +557,7 @@ class VanillaChronicleMap, createHashLookups(start); start += align64(sizeOfMultiMap() + sizeOfMultiMapBitSet()) * multiMapsPerSegment(); final NativeBytes bsBytes = new NativeBytes(tmpBytes.objectSerializer(), - start, start + ((entriesPerSegment + 7) / 8), null); + start, start + ((entriesPerSegment + 7L) / 8L), null); freeList = new SingleThreadedDirectBitSet(bsBytes); start += numberOfBitSets() * sizeOfBitSets(); entriesOffset = start - bytes.startAddr(); @@ -574,12 +568,12 @@ class VanillaChronicleMap, hashLookup = createMultiMap(start); } - public IntIntMultiMap getHashLookup() { + public MultiMap getHashLookup() { return hashLookup; } - IntIntMultiMap createMultiMap(long start) { + MultiMap createMultiMap(long start) { final NativeBytes multiMapBytes = new NativeBytes(new VanillaBytesMarshallerFactory(), start, start = start + sizeOfMultiMap(), null); @@ -606,26 +600,26 @@ class VanillaChronicleMap, * increments the size by one */ void incrementSize() { - this.bytes.addInt(SIZE_OFFSET, 1); + this.bytes.addLong(SIZE_OFFSET, 1L); } void resetSize() { - this.bytes.writeInt(SIZE_OFFSET, 0); + this.bytes.writeLong(SIZE_OFFSET, 0L); } /** * decrements the size by one */ void decrementSize() { - this.bytes.addInt(SIZE_OFFSET, -1); + this.bytes.addLong(SIZE_OFFSET, -1L); } /** * reads the the number of entries in this segment */ - int getSize() { + long getSize() { // any negative value is in error state. - return Math.max(0, this.bytes.readVolatileInt(SIZE_OFFSET)); + return Math.max(0L, this.bytes.readVolatileLong(SIZE_OFFSET)); } @@ -675,6 +669,7 @@ class VanillaChronicleMap, return entriesOffset + pos * entrySize; } + //TODO get rid of this function, integral division is slow long posFromOffset(long offset) { return (offset - entriesOffset) / entrySize; } @@ -708,9 +703,9 @@ class VanillaChronicleMap, if (sizeInBytes <= entrySize) return 1; // int division is MUCH faster than long on Intel CPUs - sizeInBytes -= 1; + sizeInBytes -= 1L; if (sizeInBytes <= Integer.MAX_VALUE) - return (((int) sizeInBytes) / entrySize) + 1; + return (((int) sizeInBytes) / (int) entrySize) + 1; return (int) (sizeInBytes / entrySize) + 1; } @@ -727,7 +722,7 @@ class VanillaChronicleMap, * this method, or {@code null}. */ V acquire(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue, - int hash2, boolean create) { + long hash2, boolean create) { if (create) writeLock(); else @@ -737,7 +732,7 @@ class VanillaChronicleMap, MultiStoreBytes entry = tmpBytes; long offset = searchKey(keyInterop, metaKeyInterop, key, keySize, hash2, entry, hashLookup); - if (offset >= 0) { + if (offset >= 0L) { return onKeyPresentOnAcquire(copies, key, usingValue, offset, entry); } else { boolean usingValuePassed = usingValue != null; @@ -765,10 +760,10 @@ class VanillaChronicleMap, } } - long searchKey(KI keyInterop, MKI metaKeyInterop, K key, long keySize, int hash2, - MultiStoreBytes entry, IntIntMultiMap hashLookup) { + long searchKey(KI keyInterop, MKI metaKeyInterop, K key, long keySize, long hash2, + MultiStoreBytes entry, MultiMap hashLookup) { hashLookup.startSearch(hash2); - for (int pos; (pos = hashLookup.nextPos()) >= 0; ) { + for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); reuse(entry, offset); if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) @@ -810,18 +805,18 @@ class VanillaChronicleMap, } } else { if (usingValue instanceof Byteable) - ((Byteable) usingValue).bytes(null, 0); + ((Byteable) usingValue).bytes(null, 0L); return defaultValueProvider.get(key, usingValue); } } V put(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V value, - int hash2, boolean replaceIfPresent) { + long hash2, boolean replaceIfPresent) { writeLock(); try { long keySize = metaKeyInterop.size(keyInterop, key); hashLookup.startSearch(hash2); - for (int pos; (pos = hashLookup.nextPos()) >= 0; ) { + for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) @@ -846,8 +841,8 @@ class VanillaChronicleMap, } } - V replaceValueOnPut(ThreadLocalCopies copies, K key, V value, NativeBytes entry, int pos, - long offset, boolean readPrevValue, IntIntMultiMap searchedHashLookup) { + V replaceValueOnPut(ThreadLocalCopies copies, K key, V value, NativeBytes entry, long pos, + long offset, boolean readPrevValue, MultiMap searchedHashLookup) { long valueSizePos = entry.position(); long valueSize = readValueSize(entry); long entryEndAddr = entry.positionAddr() + valueSize; @@ -894,7 +889,7 @@ class VanillaChronicleMap, } long entrySize = entrySize(keySize, valueSize); - int pos = alloc(inBlocks(entrySize)); + long pos = alloc(inBlocks(entrySize)); long offset = offsetFromPos(pos); clearMetaData(offset); NativeBytes entry = entry(offset); @@ -929,14 +924,13 @@ class VanillaChronicleMap, bytes.zeroOut(offset, offset + metaDataBytes); } - int alloc(int blocks) { + long alloc(int blocks) { if (blocks > MAX_ENTRY_OVERSIZE_FACTOR) throw new IllegalArgumentException("Entry is too large: requires " + blocks + " entry size chucks, " + MAX_ENTRY_OVERSIZE_FACTOR + " is maximum."); - int ret = (int) freeList.setNextNContinuousClearBits(nextPosToSearchFrom, - blocks); + long ret = freeList.setNextNContinuousClearBits(nextPosToSearchFrom, blocks); if (ret == DirectBitSet.NOT_FOUND) { - ret = (int) freeList.setNextNContinuousClearBits(0, blocks); + ret = freeList.setNextNContinuousClearBits(0L, blocks); if (ret == DirectBitSet.NOT_FOUND) { if (blocks == 1) { throw new IllegalStateException( @@ -960,12 +954,12 @@ class VanillaChronicleMap, return ret; } - private void updateNextPosToSearchFrom(int allocated, int blocks) { + private void updateNextPosToSearchFrom(long allocated, int blocks) { if ((nextPosToSearchFrom = allocated + blocks) >= freeList.size()) - nextPosToSearchFrom = 0; + nextPosToSearchFrom = 0L; } - private boolean realloc(int fromPos, int oldBlocks, int newBlocks) { + private boolean realloc(long fromPos, int oldBlocks, int newBlocks) { if (freeList.allClear(fromPos + oldBlocks, fromPos + newBlocks)) { freeList.set(fromPos + oldBlocks, fromPos + newBlocks); return true; @@ -974,7 +968,7 @@ class VanillaChronicleMap, } } - void free(int fromPos, int blocks) { + void free(long fromPos, int blocks) { freeList.clear(fromPos, fromPos + blocks); if (fromPos < nextPosToSearchFrom) nextPosToSearchFrom = fromPos; @@ -1008,12 +1002,12 @@ class VanillaChronicleMap, * exists and {@link #removeReturnsNull} is {@code false}, {@code null} otherwise */ V remove(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, - V expectedValue, int hash2) { + V expectedValue, long hash2) { writeLock(); try { long keySize = metaKeyInterop.size(keyInterop, key); hashLookup.startSearch(hash2); - for (int pos; (pos = hashLookup.nextPos()) >= 0; ) { + for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) @@ -1039,13 +1033,13 @@ class VanillaChronicleMap, } } - boolean containsKey(KI keyInterop, MKI metaKeyInterop, K key, int hash2) { + boolean containsKey(KI keyInterop, MKI metaKeyInterop, K key, long hash2) { readLock(); try { long keySize = metaKeyInterop.size(keyInterop, key); - IntIntMultiMap hashLookup = containsKeyHashLookup(); + MultiMap hashLookup = containsKeyHashLookup(); hashLookup.startSearch(hash2); - for (int pos; (pos = hashLookup.nextPos()) >= 0; ) { + for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { Bytes entry = entry(offsetFromPos(pos)); if (keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) return true; @@ -1056,7 +1050,7 @@ class VanillaChronicleMap, } } - IntIntMultiMap containsKeyHashLookup() { + MultiMap containsKeyHashLookup() { return hashLookup; } @@ -1069,12 +1063,12 @@ class VanillaChronicleMap, * @return the replaced value or {@code null} if the value was not replaced */ V replace(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, - V expectedValue, V newValue, int hash2) { + V expectedValue, V newValue, long hash2) { writeLock(); try { long keySize = metaKeyInterop.size(keyInterop, key); hashLookup.startSearch(hash2); - for (int pos; (pos = hashLookup.nextPos()) >= 0; ) { + for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) @@ -1092,8 +1086,8 @@ class VanillaChronicleMap, } V onKeyPresentOnReplace(ThreadLocalCopies copies, K key, V expectedValue, V newValue, - int pos, long offset, NativeBytes entry, - IntIntMultiMap searchedHashLookup) { + long pos, long offset, NativeBytes entry, + MultiMap searchedHashLookup) { long valueSizePos = entry.position(); long valueSize = readValueSize(entry); long entryEndAddr = entry.positionAddr() + valueSize; @@ -1104,8 +1098,7 @@ class VanillaChronicleMap, // putValue may relocate entry and change offset offset = putValue(pos, offset, entry, valueSizePos, entryEndAddr, copies, newValue, null, searchedHashLookup); - notifyPut(offset, false, key, newValue, - posFromOffset(offset)); + notifyPut(offset, false, key, newValue, posFromOffset(offset)); return valueRead; } return null; @@ -1128,7 +1121,7 @@ class VanillaChronicleMap, } } - void notifyRemoved(long offset, K key, V value, final int pos) { + void notifyRemoved(long offset, K key, V value, final long pos) { if (eventListener() != MapEventListeners.NOP) { tmpBytes.storePositionAndSize(bytes, offset, entrySize); eventListener().onRemove(VanillaChronicleMap.this, tmpBytes, metaDataBytes, @@ -1149,9 +1142,9 @@ class VanillaChronicleMap, * @return relative offset of the entry in Segment bytes after putting value (that may cause entry * relocation) */ - long putValue(int pos, long offset, NativeBytes entry, long valueSizePos, + long putValue(long pos, long offset, NativeBytes entry, long valueSizePos, long entryEndAddr, ThreadLocalCopies copies, V value, Bytes valueBytes, - IntIntMultiMap searchedHashLookup) { + MultiMap searchedHashLookup) { long valueSizeAddr = entry.address() + valueSizePos; long newValueSize; VW valueWriter = null; @@ -1188,7 +1181,7 @@ class VanillaChronicleMap, // RELOCATION free(pos, oldSizeInBlocks); eventListener().onRelocation(pos, this); - int prevPos = pos; + long prevPos = pos; pos = alloc(newSizeInBlocks); // putValue() is called from put() and replace() // after successful search by key @@ -1223,7 +1216,8 @@ class VanillaChronicleMap, return offset; } - void replacePosInHashLookupOnRelocation(IntIntMultiMap searchedHashLookup, int prevPos, int pos) { + void replacePosInHashLookupOnRelocation(MultiMap searchedHashLookup, + long prevPos, long pos) { searchedHashLookup.replacePrevPos(pos); } @@ -1232,7 +1226,7 @@ class VanillaChronicleMap, try { hashLookup.clear(); freeList.clear(); - nextPosToSearchFrom = 0; + nextPosToSearchFrom = 0L; resetSize(); } finally { writeUnlock(); @@ -1260,8 +1254,8 @@ class VanillaChronicleMap, void checkConsistency() { readLock(); try { - IntIntMultiMap hashLookup = checkConsistencyHashLookup(); - for (int pos = 0; (pos = (int) freeList.nextSetBit(pos)) >= 0; ) { + MultiMap hashLookup = checkConsistencyHashLookup(); + for (long pos = 0L; (pos = freeList.nextSetBit(pos)) >= 0L; ) { PosPresentOnce check = new PosPresentOnce(pos); hashLookup.forEach(check); if (check.count != 1) @@ -1287,19 +1281,20 @@ class VanillaChronicleMap, // no-op } - IntIntMultiMap checkConsistencyHashLookup() { + MultiMap checkConsistencyHashLookup() { return hashLookup; } - private class PosPresentOnce implements IntIntMultiMap.EntryConsumer { - int pos, count = 0; + private class PosPresentOnce implements MultiMap.EntryConsumer { + long pos; + int count = 0; - PosPresentOnce(int pos) { + PosPresentOnce(long pos) { this.pos = pos; } @Override - public void accept(int hash, int pos) { + public void accept(long hash, long pos) { if (this.pos == pos) count++; } } @@ -1367,7 +1362,7 @@ class VanillaChronicleMap, if (segIndex < 0) throw new IllegalStateException(); final Segment segment = segments[segIndex]; - final int pos = (int) returnedPos; + final long pos = returnedPos; try { segment.writeLock(); if (segment.getHashLookup().getPositions().isClear(pos)) { @@ -1389,7 +1384,7 @@ class VanillaChronicleMap, } } - void removePresent(Segment segment, int pos) { + void removePresent(Segment segment, long pos) { // TODO handle the case: // iterator.next() -- thread 1 // map.put() which cause relocation of the key, returned above -- thread 2 @@ -1404,7 +1399,7 @@ class VanillaChronicleMap, final long limit = entry.limit(); final long keySize = keySizeMarshaller.readSize(entry); long position = entry.position(); - final int segmentHash = segmentHash(Hasher.hash(entry, position, position + keySize)); + final long segmentHash = segmentHash(Hasher.hash(entry, position, position + keySize)); entry.skip(keySize); long valueSize = readValueSize(entry); @@ -1478,8 +1473,8 @@ class VanillaChronicleMap, } } - int[] segmentSizes() { - int[] sizes = new int[segments.length]; + long[] segmentSizes() { + long[] sizes = new long[segments.length]; for (int i = 0; i < segments.length; i++) sizes[i] = segments[i].getSize(); return sizes; diff --git a/src/main/java/net/openhft/chronicle/map/VanillaIntIntMultiMap.java b/src/main/java/net/openhft/chronicle/map/VanillaIntIntMultiMap.java index a10c620ee..09e1c5b80 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaIntIntMultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaIntIntMultiMap.java @@ -24,182 +24,183 @@ import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.io.Bytes; import net.openhft.lang.io.DirectStore; +import static net.openhft.lang.Maths.isPowerOf2; + /** * Supports a simple interface for int -> int[] off heap. */ -class VanillaIntIntMultiMap implements IntIntMultiMap { +class VanillaIntIntMultiMap implements MultiMap { - static int multiMapCapacity(int minCapacity) { - if (minCapacity < 0) - throw new IllegalArgumentException(); - int capacity = Maths.nextPower2(minCapacity, 16); + static final long MAX_CAPACITY = (1L << 32); + + static long multiMapCapacity(long minCapacity) { + if (minCapacity < 0L) + throw new IllegalArgumentException("minCapacity should be positive"); + long capacity = Maths.nextPower2(minCapacity, 16L); if (((double) minCapacity) / capacity > 2./3.) { // multi map shouldn't be too dense - capacity <<= 1; + capacity <<= 1L; } return capacity; } - private static final int ENTRY_SIZE = 8; + private static final long ENTRY_SIZE = 8L; private static final int ENTRY_SIZE_SHIFT = 3; - private static final int UNSET_KEY = 0; - private static final int HASH_INSTEAD_OF_UNSET_KEY = -1; - private static final int UNSET_VALUE = Integer.MIN_VALUE; - /** - * hash is in 32 higher order bits, because in Intel's little-endian they are written first - * in memory, and in memory we have keys and values in natural order: 4 bytes of k1, 4 bytes - * of v1, 4 bytes of k2, ... and this is somehow compatible with previous version of this class, - * where keys were written before values explicitly. However, this layout increases latency - * of map operations by 1 clock cycle, because we always need to perform shift to obtain the key - * between memory read and comparison with UNSET_KEY. - */ - private static final long UNSET_ENTRY = Integer.MIN_VALUE & 0xFFFFFFFFL; + private static final long UNSET_KEY = 0L; + private static final long HASH_INSTEAD_OF_UNSET_KEY = 0xFFFFFFFFL; + private static final long UNSET_ENTRY = 0L; + final DirectBitSet positions; - private final int capacity; - private final int capacityMask; + private final long capacity; + private final long capacityMask; private final long capacityMask2; private final Bytes bytes; - private int searchHash = -1; + + private long searchHash = -1L; private long searchPos = -1L; - public VanillaIntIntMultiMap(int minCapacity) { + public VanillaIntIntMultiMap(long minCapacity) { capacity = multiMapCapacity(minCapacity); - capacityMask = capacity - 1; - capacityMask2 = indexToPos(capacity - 1); + capacityMask = capacity - 1L; + capacityMask2 = indexToPos(capacity - 1L); bytes = DirectStore.allocateLazy(indexToPos(capacity)).bytes(); positions = newPositions(capacity); clear(); } public VanillaIntIntMultiMap(Bytes multiMapBytes, Bytes multiMapBitSetBytes) { - capacity = (int) (multiMapBytes.capacity() / ENTRY_SIZE); - assert capacity == Maths.nextPower2(capacity, 16); - capacityMask = capacity - 1; - capacityMask2 = indexToPos(capacity - 1); + capacity = multiMapBytes.capacity() / ENTRY_SIZE; + assert capacity == Maths.nextPower2(capacity, 16L); + capacityMask = capacity - 1L; + capacityMask2 = indexToPos(capacity - 1L); this.bytes = multiMapBytes; positions = new ATSDirectBitSet(multiMapBitSetBytes); } /** - * @param minCapacity as in {@link #VanillaIntIntMultiMap(int)} constructor + * @param minCapacity as in {@link #VanillaIntIntMultiMap(long)} constructor * @return size of {@link Bytes} to provide to {@link #VanillaIntIntMultiMap(Bytes, Bytes)} * constructor as the first argument */ - public static long sizeInBytes(int minCapacity) { + public static long sizeInBytes(long minCapacity) { return indexToPos(multiMapCapacity(minCapacity)); } /** - * @param minCapacity as in {@link #VanillaIntIntMultiMap(int)} constructor + * @param minCapacity as in {@link #VanillaIntIntMultiMap(long)} constructor * @return size of {@link Bytes} to provide to {@link #VanillaIntIntMultiMap(Bytes, Bytes)} * constructor as the second argument */ - public static long sizeOfBitSetInBytes(int minCapacity) { + public static long sizeOfBitSetInBytes(long minCapacity) { return Math.max(multiMapCapacity(minCapacity), 64L) / 8L; } - public static ATSDirectBitSet newPositions(int capacity) { + public static ATSDirectBitSet newPositions(long capacity) { + if (!isPowerOf2(capacity)) + throw new AssertionError("capacity should be a power of 2"); // bit set size should be at least 1 native long (in bits) - capacity = Math.max(capacity, 64); + capacity = Math.max(capacity, 64L); // no round-up, because capacity is already a power of 2 - int bitSetSizeInBytes = capacity / 8; + long bitSetSizeInBytes = capacity / 8L; return new ATSDirectBitSet(DirectStore.allocateLazy(bitSetSizeInBytes).bytes()); } - /** - * Separate method because it is too easy to forget to cast to long before shifting. - */ - private static long indexToPos(int index) { - return ((long) index) << ENTRY_SIZE_SHIFT; + private static long indexToPos(long index) { + return index << ENTRY_SIZE_SHIFT; } - private static int maskUnsetKey(int key) { - return key != UNSET_KEY ? key : HASH_INSTEAD_OF_UNSET_KEY; + private static long maskUnsetKey(long key) { + return (key &= 0xFFFFFFFFL) != UNSET_KEY ? key : HASH_INSTEAD_OF_UNSET_KEY; + } + + private void checkValueForPut(long value) { + assert (value & ~0xFFFFFFFFL) == 0L : "Value out of range, was " + value; + assert positions.isClear(value) : "Shouldn't put existing value"; + } + + private void checkValueForRemove(long value) { + assert (value & ~0xFFFFFFFFL) == 0L : "Value out of range, was " + value; + assert positions.isSet(value) : "Shouldn't remove absent value"; + } + + private static long key(long entry) { + return entry >>> 32; + } + + private static long value(long entry) { + return entry & 0xFFFFFFFFL; + } + + private static long entry(long key, long value) { + return (key << 32) | value; + } + + private long pos(long key) { + return indexToPos(key & capacityMask); + } + + private long step(long pos) { + return (pos + ENTRY_SIZE) & capacityMask2; + } + + private long stepBack(long pos) { + return (pos - ENTRY_SIZE) & capacityMask2; } @Override - public void put(int key, int value) { + public void put(long key, long value) { key = maskUnsetKey(key); - long pos = indexToPos(key & capacityMask); - for (int i = 0; i <= capacityMask; i++) { + checkValueForPut(value); + for (long pos = pos(key); ; pos = step(pos)) { long entry = bytes.readLong(pos); - int hash2 = (int) (entry >> 32); - if (hash2 == UNSET_KEY) { - bytes.writeLong(pos, (((long) key) << 32) | (value & 0xFFFFFFFFL)); + if (entry == UNSET_ENTRY) { + bytes.writeLong(pos, entry(key, value)); positions.set(value); return; } - if (hash2 == key) { - int value2 = (int) entry; - if (value2 == value) - return; - } - pos = (pos + ENTRY_SIZE) & capacityMask2; } - throw new IllegalStateException(getClass().getSimpleName() + " is full"); } @Override - public boolean remove(int key, int value) { + public void remove(long key, long value) { key = maskUnsetKey(key); - long pos = indexToPos(key & capacityMask); - long posToRemove = -1L; - for (int i = 0; i <= capacityMask; i++) { + checkValueForRemove(value); + long posToRemove; + for (long pos = pos(key); ; pos = step(pos)) { long entry = bytes.readLong(pos); - int hash2 = (int) (entry >> 32); - if (hash2 == key) { - int value2 = (int) entry; - if (value2 == value) { - posToRemove = pos; - break; - } - } else if (hash2 == UNSET_KEY) { + if (key(entry) == key && value(entry) == value) { + posToRemove = pos; break; } - pos = (pos + ENTRY_SIZE) & capacityMask2; } - if (posToRemove < 0) - return false; positions.clear(value); removePos(posToRemove); - return true; } - ///////////////////// - // Stateful methods - @Override - public boolean replace(int key, int oldValue, int newValue) { + public void replace(long key, long oldValue, long newValue) { key = maskUnsetKey(key); - long pos = indexToPos(key & capacityMask); - for (int i = 0; i <= capacityMask; i++) { + checkValueForRemove(oldValue); + checkValueForPut(newValue); + for (long pos = pos(key); ; pos = step(pos)) { long entry = bytes.readLong(pos); - int hash2 = (int) (entry >> 32); - if (hash2 == key) { - int value2 = (int) entry; - if (value2 == oldValue) { - positions.clear(oldValue); - positions.set(newValue); - bytes.writeLong(pos, (((long) key) << 32) | (newValue & 0xFFFFFFFFL)); - return true; - } - } else if (hash2 == UNSET_KEY) { - break; + if (key(entry) == key && value(entry) == oldValue) { + positions.clear(oldValue); + positions.set(newValue); + bytes.writeLong(pos, entry(key, newValue)); + return; } - pos = (pos + ENTRY_SIZE) & capacityMask2; } - return false; } private void removePos(long posToRemove) { long posToShift = posToRemove; - for (int i = 0; i <= capacityMask; i++) { - posToShift = (posToShift + ENTRY_SIZE) & capacityMask2; + while (true) { + posToShift = step(posToShift); long entryToShift = bytes.readLong(posToShift); - int hash = (int) (entryToShift >> 32); - if (hash == UNSET_KEY) + if (entryToShift == UNSET_ENTRY) break; - long insertPos = indexToPos(hash & capacityMask); + long insertPos = pos(key(entryToShift)); // the following condition essentially means circular permutations // of three (r = posToRemove, s = posToShift, i = insertPos) // positions are accepted: @@ -218,64 +219,61 @@ class VanillaIntIntMultiMap implements IntIntMultiMap { bytes.writeLong(posToRemove, UNSET_ENTRY); } + ///////////////////// + // Stateful methods + @Override - public int startSearch(int key) { + public void startSearch(long key) { key = maskUnsetKey(key); - searchPos = indexToPos(key & capacityMask); - return searchHash = key; + searchPos = pos(key); + searchHash = key; } @Override - public int nextPos() { + public long nextPos() { long pos = searchPos; - for (int i = 0; i < capacity; i++) { + while (true) { long entry = bytes.readLong(pos); - int hash2 = (int) (entry >> 32); - if (hash2 == UNSET_KEY) { + if (entry == UNSET_ENTRY) { searchPos = pos; - return UNSET_VALUE; + return -1L; } - pos = (pos + ENTRY_SIZE) & capacityMask2; - if (hash2 == searchHash) { + pos = step(pos); + if (key(entry) == searchHash) { searchPos = pos; - return (int) entry; + return value(entry); } } - // if return UNSET_VALUE, we have 2 cases in putAfterFailedSearch() - throw new IllegalStateException(getClass().getSimpleName() + " is full"); } @Override public void removePrevPos() { - long prevPos = (searchPos - ENTRY_SIZE) & capacityMask2; + long prevPos = stepBack(searchPos); long entry = bytes.readLong(prevPos); - int value = (int) entry; - positions.clear(value); + positions.clear(value(entry)); removePos(prevPos); } @Override - public void replacePrevPos(int newValue) { - long prevPos = ((searchPos - ENTRY_SIZE) & capacityMask2); + public void replacePrevPos(long newValue) { + checkValueForPut(newValue); + long prevPos = stepBack(searchPos); long oldEntry = bytes.readLong(prevPos); - int oldValue = (int) oldEntry; + long oldValue = value(oldEntry); positions.clear(oldValue); positions.set(newValue); - // Don't need to overwrite searchHash, but we don't know our bytes - // byte order, and can't determine offset of the value within entry. - long newEntry = (((long) searchHash) << 32) | (newValue & 0xFFFFFFFFL); - bytes.writeLong(prevPos, newEntry); + bytes.writeLong(prevPos, entry(searchHash, newValue)); } @Override - public void putAfterFailedSearch(int value) { + public void putAfterFailedSearch(long value) { + checkValueForPut(value); positions.set(value); - long entry = (((long) searchHash) << 32) | (value & 0xFFFFFFFFL); - bytes.writeLong(searchPos, entry); + bytes.writeLong(searchPos, entry(searchHash, value)); } - public int getSearchHash() { + public long getSearchHash() { return searchHash; } @@ -283,12 +281,10 @@ class VanillaIntIntMultiMap implements IntIntMultiMap { public String toString() { StringBuilder sb = new StringBuilder(); sb.append("{ "); - for (int i = 0, pos = 0; i < capacity; i++, pos += ENTRY_SIZE) { + for (long i = 0L, pos = 0L; i < capacity; i++, pos += ENTRY_SIZE) { long entry = bytes.readLong(pos); - int key = (int) (entry >> 32); - int value = (int) entry; - if (key != UNSET_KEY) - sb.append(key).append('=').append(value).append(", "); + if (entry != UNSET_ENTRY) + sb.append(key(entry)).append('=').append(value(entry)).append(", "); } if (sb.length() > 2) { sb.setLength(sb.length() - 2); @@ -299,13 +295,10 @@ class VanillaIntIntMultiMap implements IntIntMultiMap { @Override public void forEach(EntryConsumer action) { - for (int i = 0, pos = 0; i < capacity; i++, pos += ENTRY_SIZE) { + for (long i = 0L, pos = 0L; i < capacity; i++, pos += ENTRY_SIZE) { long entry = bytes.readLong(pos); - int key = (int) (entry >> 32); - int value = (int) entry; - if (key != UNSET_KEY) { - action.accept(key, value); - } + if (entry != UNSET_ENTRY) + action.accept(key(entry), value(entry)); } } @@ -317,8 +310,6 @@ class VanillaIntIntMultiMap implements IntIntMultiMap { @Override public void clear() { positions.clear(); - for (int pos = 0; pos < bytes.capacity(); pos += ENTRY_SIZE) { - bytes.writeLong(pos, UNSET_ENTRY); - } + bytes.zeroOut(); } } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaShortShortMultiMap.java b/src/main/java/net/openhft/chronicle/map/VanillaShortShortMultiMap.java index 4e7526985..e860c33ea 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaShortShortMultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaShortShortMultiMap.java @@ -19,177 +19,168 @@ package net.openhft.chronicle.map; -import net.openhft.lang.Maths; import net.openhft.lang.collection.ATSDirectBitSet; import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.io.Bytes; import net.openhft.lang.io.DirectStore; import static net.openhft.chronicle.map.VanillaIntIntMultiMap.multiMapCapacity; +import static net.openhft.lang.Maths.isPowerOf2; /** * Supports a simple interface for int -> int[] off heap. */ -class VanillaShortShortMultiMap implements IntIntMultiMap { +class VanillaShortShortMultiMap implements MultiMap { - private static final int ENTRY_SIZE = 4; + static final long MAX_CAPACITY = (1L << 16); + + private static final long ENTRY_SIZE = 4L; private static final int ENTRY_SIZE_SHIFT = 2; private static final int UNSET_KEY = 0; - private static final int HASH_INSTEAD_OF_UNSET_KEY = 0xFFFF; - private static final int UNSET_VALUE = Integer.MIN_VALUE; - private static final int UNSET_ENTRY = 0xFFFF; + private static final long HASH_INSTEAD_OF_UNSET_KEY = 0xFFFFL; + private static final int UNSET_ENTRY = 0; private final int capacity; - private final int capacityMask; - private final int capacityMask2; + private final long capacityMask; + private final long capacityMask2; private final Bytes bytes; private ATSDirectBitSet positions; - private int searchHash = -1; - private int searchPos = -1; + private long searchHash = -1L; + private long searchPos = -1L; - public VanillaShortShortMultiMap(int minCapacity) { - capacity = multiMapCapacity(minCapacity); - if (capacity > (1 << 16)) - throw new IllegalArgumentException(); - capacityMask = capacity - 1; - capacityMask2 = (capacity - 1) * ENTRY_SIZE; + public VanillaShortShortMultiMap(long minCapacity) { + assert minCapacity <= MAX_CAPACITY; + long capacity = multiMapCapacity(minCapacity); + assert isPowerOf2(capacity); + this.capacity = (int) capacity; + capacityMask = capacity - 1L; + capacityMask2 = (capacity - 1L) * ENTRY_SIZE; bytes = DirectStore.allocateLazy(capacity * ENTRY_SIZE).bytes(); positions = VanillaIntIntMultiMap.newPositions(capacity); clear(); } public VanillaShortShortMultiMap(Bytes multiMapBytes, Bytes multiMapBitSetBytes) { - capacity = (int) (multiMapBytes.capacity() / ENTRY_SIZE); - assert capacity == Maths.nextPower2(capacity, 16); - capacityMask = capacity - 1; - capacityMask2 = (capacity - 1) * ENTRY_SIZE; + long capacity = multiMapBytes.capacity() / ENTRY_SIZE; + assert isPowerOf2(capacity); + assert (capacity / 2L) < MAX_CAPACITY; + this.capacity = (int) capacity; + capacityMask = capacity - 1L; + capacityMask2 = (capacity - 1L) * ENTRY_SIZE; this.bytes = multiMapBytes; positions = new ATSDirectBitSet(multiMapBitSetBytes); } /** - * @param minCapacity as in {@link #VanillaShortShortMultiMap(int)} constructor + * @param minCapacity as in {@link #VanillaShortShortMultiMap(long)} constructor * @return size of {@link Bytes} to provide to {@link #VanillaShortShortMultiMap(Bytes, Bytes)} * constructor as the first argument */ - public static long sizeInBytes(int minCapacity) { - return ((long) multiMapCapacity(minCapacity)) * ENTRY_SIZE; + public static long sizeInBytes(long minCapacity) { + return multiMapCapacity(minCapacity) * ENTRY_SIZE; } /** - * @param minCapacity as in {@link #VanillaShortShortMultiMap(int)} constructor + * @param minCapacity as in {@link #VanillaShortShortMultiMap(long)} constructor * @return size of {@link Bytes} to provide to {@link #VanillaShortShortMultiMap(Bytes, Bytes)} * constructor as the second argument */ - public static long sizeOfBitSetInBytes(int minCapacity) { + public static long sizeOfBitSetInBytes(long minCapacity) { return VanillaIntIntMultiMap.sizeOfBitSetInBytes(minCapacity); } - private static void checkKey(int key) { - if ((key & ~0xFFFF) != 0) - throw new IllegalArgumentException("Key out of range, was " + key); + private void checkValueForPut(long value) { + assert (value & ~0xFFFFL) == 0L : "Value out of range, was " + value; + assert positions.isClear(value) : "Shouldn't put existing value"; } - private static void checkValue(int value) { - if ((value & ~0xFFFF) != 0) - throw new IllegalArgumentException("Value out of range, was " + value); + private void checkValueForRemove(long value) { + assert (value & ~0xFFFFL) == 0L : "Value out of range, was " + value; + assert positions.isSet(value) : "Shouldn't remove absent value"; } - private static int checkAndMaskUnsetKey(int key) { - if (key == UNSET_KEY) { - return HASH_INSTEAD_OF_UNSET_KEY; - } else { - checkKey(key); - return key; - } + + private static long maskUnsetKey(long key) { + return (key &= 0xFFFFL) != UNSET_KEY ? key : HASH_INSTEAD_OF_UNSET_KEY; + } + + private long step(long pos) { + return (pos + ENTRY_SIZE) & capacityMask2; + } + + private long stepBack(long pos) { + return (pos - ENTRY_SIZE) & capacityMask2; + } + + private long pos(long key) { + return (key & capacityMask) << ENTRY_SIZE_SHIFT; + } + + private static int entry(long key, long value) { + return (int) ((key << 16) | value); + } + + private static long key(int entry) { + return (long) (entry >>> 16); + } + + private static long value(int entry) { + return (long) (entry & 0xFFFF); } @Override - public void put(int key, int value) { - key = checkAndMaskUnsetKey(key); - checkValue(value); - int pos = (key & capacityMask) << ENTRY_SIZE_SHIFT; - for (int i = 0; i <= capacityMask; i++) { + public void put(long key, long value) { + key = maskUnsetKey(key); + checkValueForPut(value); + for (long pos = pos(key); ; pos = step(pos)) { int entry = bytes.readInt(pos); - int hash2 = entry >>> 16; - if (hash2 == UNSET_KEY) { - bytes.writeInt(pos, ((key << 16) | value)); + if (entry == UNSET_ENTRY) { + bytes.writeInt(pos, entry(key, value)); positions.set(value); return; } - if (hash2 == key) { - int value2 = entry & 0xFFFF; - if (value2 == value) - return; - } - pos = (pos + ENTRY_SIZE) & capacityMask2; } - throw new IllegalStateException(getClass().getSimpleName() + " is full"); } @Override - public boolean remove(int key, int value) { - key = checkAndMaskUnsetKey(key); - checkValue(value); - int pos = (key & capacityMask) << ENTRY_SIZE_SHIFT; - int posToRemove = -1; - for (int i = 0; i <= capacityMask; i++) { + public void remove(long key, long value) { + key = maskUnsetKey(key); + checkValueForRemove(value); + long posToRemove; + for (long pos = pos(key); ; pos = step(pos)) { int entry = bytes.readInt(pos); - int hash2 = entry >>> 16; - if (hash2 == key) { - int value2 = entry & 0xFFFF; - if (value2 == value) { - posToRemove = pos; - break; - } - } else if (hash2 == UNSET_KEY) { + if (key(entry) == key && value(entry) == value) { + posToRemove = pos; break; } - pos = (pos + ENTRY_SIZE) & capacityMask2; } - if (posToRemove < 0) - return false; positions.clear(value); removePos(posToRemove); - return true; } - ///////////////////// - // Stateful methods - @Override - public boolean replace(int key, int oldValue, int newValue) { - key = checkAndMaskUnsetKey(key); - checkValue(oldValue); - checkValue(newValue); - int pos = (key & capacityMask) << ENTRY_SIZE_SHIFT; - for (int i = 0; i <= capacityMask; i++) { + public void replace(long key, long oldValue, long newValue) { + key = maskUnsetKey(key); + checkValueForRemove(oldValue); + checkValueForPut(newValue); + for (long pos = pos(key); ; pos = step(pos)) { int entry = bytes.readInt(pos); - int hash2 = entry >>> 16; - if (hash2 == key) { - int value2 = entry & 0xFFFF; - if (value2 == oldValue) { - positions.clear(oldValue); - positions.set(newValue); - bytes.writeInt(pos, ((key << 16) | newValue)); - return true; - } - } else if (hash2 == UNSET_KEY) { - break; + if (key(entry) == key && value(entry) == oldValue) { + positions.clear(oldValue); + positions.set(newValue); + bytes.writeInt(pos, entry(key, newValue)); + return; } - pos = (pos + ENTRY_SIZE) & capacityMask2; } - return false; } - private void removePos(int posToRemove) { - int posToShift = posToRemove; - for (int i = 0; i <= capacityMask; i++) { - posToShift = (posToShift + ENTRY_SIZE) & capacityMask2; + private void removePos(long posToRemove) { + long posToShift = posToRemove; + while (true) { + posToShift = step(posToShift); int entryToShift = bytes.readInt(posToShift); - int hash = entryToShift >>> 16; - if (hash == UNSET_KEY) + if (entryToShift == UNSET_ENTRY) break; - int insertPos = (hash & capacityMask) << ENTRY_SIZE_SHIFT; + long insertPos = pos(key(entryToShift)); // see comment in VanillaIntIntMultiMap boolean cond1 = insertPos <= posToRemove; boolean cond2 = posToRemove <= posToShift; @@ -203,63 +194,60 @@ class VanillaShortShortMultiMap implements IntIntMultiMap { bytes.writeInt(posToRemove, UNSET_ENTRY); } + ///////////////////// + // Stateful methods + @Override - public int startSearch(int key) { - key = checkAndMaskUnsetKey(key); - searchPos = (key & capacityMask) << ENTRY_SIZE_SHIFT; - return searchHash = key; + public void startSearch(long key) { + key = maskUnsetKey(key); + searchPos = pos(key); + searchHash = key; } @Override - public int nextPos() { - int pos = searchPos; - for (int i = 0; i < capacity; i++) { + public long nextPos() { + long pos = searchPos; + while (true) { int entry = bytes.readInt(pos); - int hash2 = entry >>> 16; - if (hash2 == UNSET_KEY) { + if (entry == UNSET_ENTRY) { searchPos = pos; - return UNSET_VALUE; + return -1L; } - pos = (pos + ENTRY_SIZE) & capacityMask2; - if (hash2 == searchHash) { + pos = step(pos); + if (key(entry) == searchHash) { searchPos = pos; - return entry & 0xFFFF; + return value(entry); } } - // if return UNSET_VALUE, we have 2 cases in putAfterFailedSearch() - throw new IllegalStateException(getClass().getSimpleName() + " is full"); } @Override public void removePrevPos() { - int prevPos = (searchPos - ENTRY_SIZE) & capacityMask2; + long prevPos = stepBack(searchPos); int entry = bytes.readInt(prevPos); - int value = entry & 0xFFFF; - positions.clear(value); + positions.clear(value(entry)); removePos(prevPos); } @Override - public void replacePrevPos(int newValue) { - checkValue(newValue); - int prevPos = ((searchPos - ENTRY_SIZE) & capacityMask2); + public void replacePrevPos(long newValue) { + checkValueForPut(newValue); + long prevPos = stepBack(searchPos); int oldEntry = bytes.readInt(prevPos); - int oldValue = oldEntry & 0xFFFF; + long oldValue = value(oldEntry); positions.clear(oldValue); positions.set(newValue); - // Don't need to overwrite searchHash, but we don't know our bytes - // byte order, and can't determine offset of the value within entry. - bytes.writeInt(prevPos, searchHash << 16 | newValue); + bytes.writeInt(prevPos, entry(searchHash, newValue)); } @Override - public void putAfterFailedSearch(int value) { - checkValue(value); + public void putAfterFailedSearch(long value) { + checkValueForPut(value); positions.set(value); - bytes.writeInt(searchPos, searchHash << 16 | value); + bytes.writeInt(searchPos, entry(searchHash, value)); } - public int getSearchHash() { + public long getSearchHash() { return searchHash; } @@ -267,12 +255,12 @@ class VanillaShortShortMultiMap implements IntIntMultiMap { public String toString() { StringBuilder sb = new StringBuilder(); sb.append("{ "); - for (int i = 0, pos = 0; i < capacity; i++, pos += ENTRY_SIZE) { + long pos = 0L; + for (int i = 0; i < capacity; i++, pos += ENTRY_SIZE) { int entry = bytes.readInt(pos); - int key = entry >>> 16; - int value = entry & 0xFFFF; - if (key != UNSET_KEY) - sb.append(key).append('=').append(value).append(", "); + if (entry != UNSET_ENTRY) + sb.append(key(entry)).append('=').append(value(entry)).append(", "); + } if (sb.length() > 2) { sb.setLength(sb.length() - 2); @@ -283,12 +271,11 @@ class VanillaShortShortMultiMap implements IntIntMultiMap { @Override public void forEach(EntryConsumer action) { - for (int i = 0, pos = 0; i < capacity; i++, pos += ENTRY_SIZE) { + long pos = 0L; + for (int i = 0; i < capacity; i++, pos += ENTRY_SIZE) { int entry = bytes.readInt(pos); - int key = entry >>> 16; - int value = entry & 0xFFFF; - if (key != UNSET_KEY) - action.accept(key, value); + if (entry != UNSET_ENTRY) + action.accept(key(entry), value(entry)); } } @@ -300,8 +287,6 @@ class VanillaShortShortMultiMap implements IntIntMultiMap { @Override public void clear() { positions.clear(); - for (int pos = 0; pos < bytes.capacity(); pos += ENTRY_SIZE) { - bytes.writeInt(pos, UNSET_ENTRY); - } + bytes.zeroOut(); } } diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index d841202fb..e81221168 100644 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -85,7 +85,7 @@ public class ChronicleSetBuilder } @Override - public ChronicleSetBuilder actualEntriesPerSegment(int actualEntriesPerSegment) { + public ChronicleSetBuilder actualEntriesPerSegment(long actualEntriesPerSegment) { chronicleMapBuilder.actualEntriesPerSegment(actualEntriesPerSegment); return this; } @@ -164,12 +164,6 @@ public class ChronicleSetBuilder return this; } - @Override - public ChronicleSetBuilder largeSegments(boolean largeSegments) { - chronicleMapBuilder.largeSegments(largeSegments); - return this; - } - @Override public ChronicleSetBuilder metaDataBytes(int metaDataBytes) { chronicleMapBuilder.metaDataBytes(metaDataBytes); diff --git a/src/test/java/eg/BigData.java b/src/test/java/eg/BigData.java index 423b014f5..745998a78 100644 --- a/src/test/java/eg/BigData.java +++ b/src/test/java/eg/BigData.java @@ -45,7 +45,6 @@ public class BigData { // static AtomicInteger Highwatermark = new AtomicInteger(0); static { - builder.largeSegments(true); builder.actualSegments(8 * 1024); builder.entries(MAXSIZE); String dir = System.getProperty("dir", "/ocz/tmp"); diff --git a/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java b/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java index 66f884ec4..21204db14 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java +++ b/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java @@ -91,7 +91,7 @@ public class CHMMetaDataTest { @Override public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes, - String key, String value, int pos, SharedSegment segment) { + String key, String value, long pos, SharedSegment segment) { assertEquals(8, metaDataBytes); System.out.println("Removed " + key + "/" + value + " with ts of " + entry.readLong(0)); } diff --git a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java index 11366878f..c73d34b65 100644 --- a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java @@ -144,7 +144,7 @@ public class EntryCountMapTest { map.acquireUsing(key, longValue); longValue.setValue(1); } - int[] a = ((VanillaChronicleMap) map).segmentSizes(); + long[] a = ((VanillaChronicleMap) map).segmentSizes(); System.out.println("segs: " + segments + " min: " + minSize + " was " + map.size() + " " + Arrays.toString(a) + " sum: " + sum(a)); @@ -155,7 +155,7 @@ public class EntryCountMapTest { scoreCount++; boolean condition = minSize <= map.size() && map.size() <= maxSize; if (!condition) { - int[] a = ((VanillaChronicleMap) map).segmentSizes(); + long[] a = ((VanillaChronicleMap) map).segmentSizes(); System.out.println("segs: " + segments + " min: " + minSize + " was " + map.size() + " " + Arrays.toString(a) + " sum: " + sum(a)); @@ -166,9 +166,9 @@ public class EntryCountMapTest { } } - private long sum(int[] ints) { + private long sum(long[] longs) { long sum = 0; - for (int i : ints) { + for (long i : longs) { sum += i; } return sum; diff --git a/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java b/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java index f1e48373d..e43f8a312 100644 --- a/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java @@ -21,10 +21,13 @@ package net.openhft.chronicle.map; import com.google.common.collect.HashMultimap; import com.google.common.collect.Multimap; +import net.openhft.lang.collection.SingleThreadedDirectBitSet; +import net.openhft.lang.io.DirectStore; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; +import java.lang.reflect.Field; import java.util.*; import static org.junit.Assert.assertEquals; @@ -36,11 +39,11 @@ import static org.junit.Assert.assertTrue; @RunWith(value = Parameterized.class) public class IntIntMultiMapTest { - IntIntMultiMap map; - Multimap referenceMap = HashMultimap.create(); - private Class c; + MultiMap map; + Multimap referenceMap = HashMultimap.create(); + private Class c; - public IntIntMultiMapTest(Class c) + public IntIntMultiMapTest(Class c) throws Exception { this.c = c; } @@ -53,20 +56,20 @@ public class IntIntMultiMapTest { }); } - private void initMap(int capacity) { + private void initMap(long capacity) { try { - map = c.getConstructor(int.class).newInstance(capacity); + map = c.getConstructor(long.class).newInstance(capacity); } catch (Exception e) { throw new RuntimeException(e); } } private void multiMapEquals() { - class Action implements IntIntMultiMap.EntryConsumer { + class Action implements MultiMap.EntryConsumer { int mapSize = 0; @Override - public void accept(int key, int value) { + public void accept(long key, long value) { mapSize++; assertTrue(referenceMap.containsEntry(key, value)); } @@ -76,24 +79,24 @@ public class IntIntMultiMapTest { assertEquals(referenceMap.size(), action.mapSize); } - private void valuesEqualsByKey(int k) { - List values = new ArrayList(); + private void valuesEqualsByKey(long k) { + List values = new ArrayList(); map.startSearch(k); - int v; - while ((v = map.nextPos()) >= 0) + long v; + while ((v = map.nextPos()) >= 0L) values.add(v); - Set valueSet = new HashSet(values); + Set valueSet = new HashSet(values); assertEquals(values.size(), valueSet.size()); - assertEquals(new HashSet(referenceMap.get(k)), valueSet); + assertEquals(new HashSet(referenceMap.get(k)), valueSet); } - private void put(int k, int v) { + private void put(long k, long v) { map.put(k, v); referenceMap.put(k, v); } - private void remove(int k, int v, boolean present) { - assertEquals(present, map.remove(k, v)); + private void remove(long k, long v) { + map.remove(k, v); referenceMap.remove(k, v); } @@ -114,23 +117,22 @@ public class IntIntMultiMapTest { put(1, 15); multiMapEquals(); - remove(1, 11, true); + remove(1, 11); multiMapEquals(); - remove(1, 11, false); valuesEqualsByKey(3); valuesEqualsByKey(1); - remove(1, 12, true); + remove(1, 12); multiMapEquals(); - remove(1, 15, true); + remove(1, 15); multiMapEquals(); - remove(1, 13, true); + remove(1, 13); multiMapEquals(); - remove(1, 14, true); + remove(1, 14); multiMapEquals(); } @@ -146,4 +148,33 @@ public class IntIntMultiMapTest { map.startSearch(15); assertTrue(map.nextPos() < 0); } + + @Test + public void testMaxCapacity() throws NoSuchFieldException, IllegalAccessException { + Field maxCapacityField = c.getDeclaredField("MAX_CAPACITY"); + long maxCapacity = maxCapacityField.getLong(null); + Field entrySizeField = c.getDeclaredField("ENTRY_SIZE"); + entrySizeField.setAccessible(true); + long entrySize = entrySizeField.getLong(null); + if (Runtime.getRuntime().maxMemory() < maxCapacity * entrySize * 3L / 2L) { + System.out.println("Skipped " + c + " maxCapacity test because there is not enough " + + "memory in system"); + return; + } + initMap(maxCapacity); + Random r = new Random(); + int bound = maxCapacity > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) maxCapacity; + for (long i = 0L; i < maxCapacity; i++) { + map.put(r.nextInt(bound), i); + } + final SingleThreadedDirectBitSet bs = + new SingleThreadedDirectBitSet(new DirectStore(maxCapacity / 8).bytes()); + map.forEach(new MultiMap.EntryConsumer() { + @Override + public void accept(long key, long value) { + bs.set(value); + } + }); + assertTrue(bs.allSet(0L, maxCapacity)); + } }