Merge remote-tracking branch 'origin/master'

Conflicts:
	src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
This commit is contained in:
Peter Lawrey
2014-10-27 10:06:31 +00:00
18 changed files with 752 additions and 534 deletions
@@ -139,7 +139,7 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* number of chunks could be taken by an entry is 64. {@link IllegalArgumentException} is thrown
* on attempt to insert too large entry, compared to the configured or computed entry size.</li>
* <li>The number of "entries" {@linkplain #entries(long) in the whole map} and {@linkplain
* #actualEntriesPerSegment(int) per segment} is actually the number "chunks".</li> </ul>
* #actualEntriesPerSegment(long) per segment} is actually the number "chunks".</li> </ul>
*
* <p>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<K, C extends ChronicleHash,
*/
B entries(long entries);
B actualEntriesPerSegment(int actualEntriesPerSegment);
B actualEntriesPerSegment(long actualEntriesPerSegment);
B actualSegments(int actualSegments);
@@ -219,8 +219,6 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
B errorListener(ChronicleHashErrorListener errorListener);
B largeSegments(boolean largeSegments);
B metaDataBytes(int metaDataBytes);
/**
@@ -94,7 +94,10 @@ public class ChronicleMapBuilder<K, V> 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<K, V> 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<K, V> implements Cloneable,
}
}
@Override
public ChronicleMapBuilder<K, V> 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<K, V> implements Cloneable,
@Override
public ChronicleMapBuilder<K, V> entries(long entries) {
if (entries <= 0L)
throw new IllegalArgumentException("Entries should be positive, " + entries + " given");
this.entries = entries;
return this;
}
@@ -386,43 +372,114 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
}
@Override
public ChronicleMapBuilder<K, V> actualEntriesPerSegment(int actualEntriesPerSegment) {
public ChronicleMapBuilder<K, V> 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<K, V> minSegments(int minSegments) {
checkSegments(minSegments);
this.minSegments = minSegments;
return this;
}
int minSegments() {
return Math.max(estimateSegments(), minSegments);
}
private int estimateSegments() {
for (int power = 0; power < 7; power++) {
if (entries < (1L << (7 + (2 * power))))
return 1 << power;
}
// Heuristic -- number of segments ~= cube root from number of entries seems optimal
int maxSegments = 65536;
for (int segments = 4; segments < maxSegments; segments <<= 1) {
if (((long) segments) * segments * segments >= entries)
return segments;
}
return maxSegments;
}
@Override
public ChronicleMapBuilder<K, V> 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<K, V> lockTimeOut(long lockTimeOut, TimeUnit unit) {
@@ -497,16 +554,6 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
return removeReturnsNull;
}
boolean largeSegments() {
return entries > 1L << (20 + 15) || largeSegments;
}
@Override
public ChronicleMapBuilder<K, V> largeSegments(boolean largeSegments) {
this.largeSegments = largeSegments;
return this;
}
@Override
public ChronicleMapBuilder<K, V> metaDataBytes(int metaDataBytes) {
if ((metaDataBytes & 0xFF) != metaDataBytes)
@@ -538,7 +585,6 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
", errorListener=" + errorListener() +
", putReturnsNull=" + putReturnsNull() +
", removeReturnsNull=" + removeReturnsNull() +
", largeSegments=" + (largeSegments ? "true" : "not configured") +
", timeProvider=" + timeProvider() +
", bytesMarshallerFactory=" + pretty(bytesMarshallerFactory) +
", objectSerializer=" + pretty(objectSerializer) +
@@ -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;
}
}
}
@@ -95,7 +95,7 @@ public abstract class MapEventListener<K, V, M extends ChronicleMap<K, V>>
}
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<K, V, M extends ChronicleMap<K, V>>
// do nothing
}
void onRelocation(int pos, SharedSegment segment) {
void onRelocation(long pos, SharedSegment segment) {
// do nothing
}
}
@@ -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);
}
};
@@ -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);
}
}
@@ -139,12 +139,13 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
@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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
if (!isDeleted) {
valueSize = valueSizeMarshaller.readSize(entry);
} else {
valueSize = 0;
valueSize = 0L;
}
final long valuePosition = entry.position();
@@ -476,7 +477,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
class Segment extends VanillaChronicleMap<K, KI, MKI, V, VW, MVW>.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<K, KI, MKI extends MetaBytesInterop<K, KI>,
hashLookupLiveOnly = createMultiMap(start);
}
public IntIntMultiMap getHashLookup() {
public MultiMap getHashLookup() {
return hashLookupLiveOnly;
}
@@ -613,7 +614,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
/**
* 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
// 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
/**
* 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
// 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @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<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @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<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @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<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
@Override
IntIntMultiMap containsKeyHashLookup() {
MultiMap containsKeyHashLookup() {
return hashLookupLiveOnly;
}
@@ -1041,12 +1042,12 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @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<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
@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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
class EntryIterator extends VanillaChronicleMap<K, KI, MKI, V, VW, MVW>.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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
@Override
public void onRemove(ChronicleMap<K, V> 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
@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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
@Override
public void onRemove(ChronicleMap<K, V> 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
* 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
*/
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @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);
}
@@ -49,7 +49,7 @@ import static java.lang.Thread.currentThread;
class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
V, VW, MVW extends MetaBytesWriter<V, VW>> extends AbstractMap<K, V>
implements ChronicleMap<K, V>, 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, V, ChronicleMap<K, V>> 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
transient long headerSize;
transient Set<Map.Entry<K, V>> entrySet;
private int bits;
private int mask;
private final HashSplitting hashSplitting;
public VanillaChronicleMap(ChronicleMapBuilder<K, V> builder) throws IOException {
@@ -135,34 +134,32 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
this.removeReturnsNull = builder.removeReturnsNull();
this.actualSegments = builder.actualSegments();
// align by 8 because otherwise sizeOfBitSets() might address less blocks than specified
this.entriesPerSegment = align8(builder.actualEntriesPerSegment());
this.entriesPerSegment = builder.actualEntriesPerSegment();
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 +249,11 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
boolean useSmallMultiMaps() {
return entriesPerSegment <= (1 << 16);
return entriesPerSegment <= VanillaShortShortMultiMap.MAX_CAPACITY;
}
long sizeOfBitSets() {
return align64(entriesPerSegment / 8);
return align64(align8(entriesPerSegment) / 8L);
}
int numberOfBitSets() {
@@ -270,7 +267,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
+ 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 +293,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
private long sizeOfEntriesInSegment() {
return align64((long) entriesPerSegment * entrySize);
return align64(entriesPerSegment * entrySize);
}
@Override
@@ -347,7 +344,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +373,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +389,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +444,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +469,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
@Override
public long longSize() {
long result = 0;
long result = 0L;
for (final Segment segment : this.segments) {
result += segment.getSize();
@@ -496,8 +493,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @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 +503,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +533,19 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
- 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) {
@@ -562,7 +555,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
long start = bytes.startAddr() + SEGMENT_HEADER;
createHashLookups(start);
start += align64(sizeOfMultiMap() + sizeOfMultiMapBitSet()) * multiMapsPerSegment();
final NativeBytes bsBytes = new NativeBytes(ms.objectSerializer(),
final NativeBytes bsBytes = new NativeBytes(tmpBytes.objectSerializer(),
start, start + ((entriesPerSegment + 7) / 8), null);
freeList = new SingleThreadedDirectBitSet(bsBytes);
start += numberOfBitSets() * sizeOfBitSets();
@@ -574,12 +567,12 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +599,26 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* 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 +668,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return entriesOffset + pos * entrySize;
}
//TODO get rid of this function, integral division is slow
long posFromOffset(long offset) {
return (offset - entriesOffset) / entrySize;
}
@@ -708,9 +702,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +721,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* 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 +731,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +759,10 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
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 +804,18 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
} 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 +840,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
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 +888,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
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,15 +923,19 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
bytes.zeroOut(offset, offset + metaDataBytes);
}
int alloc(int blocks) {
//TODO refactor/optimize
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);
if (ret == DirectBitSet.NOT_FOUND) {
ret = (int) freeList.setNextNContinuousClearBits(0, blocks);
if (ret == DirectBitSet.NOT_FOUND) {
long ret = freeList.setNextNContinuousClearBits(nextPosToSearchFrom, blocks);
if (ret == DirectBitSet.NOT_FOUND || ret + blocks > entriesPerSegment) {
if (ret + blocks > entriesPerSegment)
freeList.clear(ret, ret + blocks);
ret = freeList.setNextNContinuousClearBits(0L, blocks);
if (ret == DirectBitSet.NOT_FOUND || ret + blocks > entriesPerSegment) {
if (ret + blocks > entriesPerSegment)
freeList.clear(ret, ret + blocks);
if (blocks == 1) {
throw new IllegalStateException(
"Segment is full, no free entries found");
@@ -960,12 +958,12 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return ret;
}
private void updateNextPosToSearchFrom(int allocated, int blocks) {
if ((nextPosToSearchFrom = allocated + blocks) >= freeList.size())
nextPosToSearchFrom = 0;
private void updateNextPosToSearchFrom(long allocated, int blocks) {
if ((nextPosToSearchFrom = allocated + blocks) >= entriesPerSegment)
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 +972,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
void free(int fromPos, int blocks) {
void free(long fromPos, int blocks) {
freeList.clear(fromPos, fromPos + blocks);
if (fromPos < nextPosToSearchFrom)
nextPosToSearchFrom = fromPos;
@@ -1008,12 +1006,12 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* 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 +1037,13 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
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 +1054,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
IntIntMultiMap containsKeyHashLookup() {
MultiMap containsKeyHashLookup() {
return hashLookup;
}
@@ -1069,12 +1067,12 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @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 +1090,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
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 +1102,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
// 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 +1125,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
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 +1146,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @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 +1185,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
// 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 +1220,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return offset;
}
void replacePosInHashLookupOnRelocation(IntIntMultiMap searchedHashLookup, int prevPos, int pos) {
void replacePosInHashLookupOnRelocation(MultiMap searchedHashLookup,
long prevPos, long pos) {
searchedHashLookup.replacePrevPos(pos);
}
@@ -1232,7 +1230,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
try {
hashLookup.clear();
freeList.clear();
nextPosToSearchFrom = 0;
nextPosToSearchFrom = 0L;
resetSize();
} finally {
writeUnlock();
@@ -1260,8 +1258,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +1285,20 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
// 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 +1366,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +1388,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
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 +1403,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
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 +1477,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
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;
@@ -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();
}
}
@@ -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();
}
}
@@ -85,7 +85,7 @@ public class ChronicleSetBuilder<E>
}
@Override
public ChronicleSetBuilder<E> actualEntriesPerSegment(int actualEntriesPerSegment) {
public ChronicleSetBuilder<E> actualEntriesPerSegment(long actualEntriesPerSegment) {
chronicleMapBuilder.actualEntriesPerSegment(actualEntriesPerSegment);
return this;
}
@@ -164,12 +164,6 @@ public class ChronicleSetBuilder<E>
return this;
}
@Override
public ChronicleSetBuilder<E> largeSegments(boolean largeSegments) {
chronicleMapBuilder.largeSegments(largeSegments);
return this;
}
@Override
public ChronicleSetBuilder<E> metaDataBytes(int metaDataBytes) {
chronicleMapBuilder.metaDataBytes(metaDataBytes);