Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Ensure there is capacity in the hash table before attempting to add an entry.

Browse Source
This commit is contained in:
Peter Lawrey
2015-12-17 12:13:49 -06:00
parent bd64095042
commit a07fd3ba3d
7 changed files with 130 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pom.xml
+1 -1
View File
@@ -44,7 +44,7 @@
<dependency>
<groupId>net.openhft</groupId>
<artifactId>chronicle-bom</artifactId>
<version>1.10.19</version>
<version>1.10.28-SNAPSHOT</version>
<type>pom</type>
<scope>import</scope>
</dependency>
src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java
Regular → Executable
+16
View File
@@ -286,6 +286,22 @@ class Int24Int24MultiMap implements MultiMap {
}
}
public long size() {
long pos = 0;
int size = 0;
for (int count = capacity; count > 0; count--) {
long entry = bytes.readInt48(pos);
pos = step(pos);
if (entry != UNSET_ENTRY)
size++;
}
return size;
}
@Override
public long capacity() {
return capacity;
}
@Override
public DirectBitSet getPositions() {
return positions;
src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java
Regular → Executable
+17
View File
@@ -285,6 +285,23 @@ class IntIntMultiMap implements MultiMap {
}
}
public long size() {
long pos = 0;
long size = 0;
for (long count = capacity; count > 0; count--) {
long entry = bytes.readLong(pos);
pos = step(pos);
if (entry != UNSET_ENTRY)
size++;
}
return size;
}
@Override
public long capacity() {
return capacity;
}
@Override
public DirectBitSet getPositions() {
return positions;
src/main/java/net/openhft/chronicle/map/MultiMap.java
Regular → Executable
+4
View File
@@ -73,6 +73,10 @@ interface MultiMap {
void forEach(EntryConsumer action);
long size();
long capacity();
interface EntryConsumer {
void accept(long key, long value);
}
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
Regular → Executable
+21 -4
View File
@@ -468,6 +468,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
public boolean onEntry(@NotNull final Bytes entry, final int chronicleId, final long bootStrapTimeStamp) {
final long keySize = keySizeMarshaller.readSize(entry);
if (keySize == 0)
System.out.println("read: " + Long.toHexString(entry.address()) + "\n" + entry.toHexString(32));
final long keyPosition = entry.position();
entry.skip(keySize);
@@ -488,8 +490,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
final NativeBytes k = wrap(entry.address() + keyPosition, keySize);
final NativeBytes v = wrap(entry.address() + valuePosition, valueSize);
assert k.remaining() > 0 : "keySize:" + keySize;
assert v.remaining() > 0 : "valueSize:" + valueSize;
assert k.remaining() > 0 : "keySize:" + keySize + " bytes: " + entry.toDebugString(128);
assert v.remaining() > 0 : "valueSize:" + valueSize + " bytes: " + entry.toDebugString(128);
return callback.onEntry(k, v,
timestamp,
@@ -875,6 +877,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
long hash2, boolean create, MutableLockedEntry lock) {
MultiStoreBytes entry = segmentState.tmpBytes;
MultiMap hashLookup = hashLookup();
System.out.println("Before put size: " + hashLookup.size());
SearchState searchState = segmentState.searchState;
hashLookup.startSearch(hash2, searchState);
for (long pos; (pos = hashLookup.nextPos(searchState)) >= 0L; ) {
@@ -932,6 +935,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
throw defaultValueOrPrepareBytesShouldBeSpecified();
}
checkPreincrementSize();
final long elemSize = metaElemWriter.size(elemWriter, elem);
putValue(pos, entry, valueSizePos,
@@ -1109,6 +1113,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
segmentState.timestamp = timeStamp;
try {
MultiMap hashLookup = hashLookup();
System.out.println("Before update size: " + hashLookup.size() + " of " + hashLookup.capacity());
SearchState searchState = segmentState.searchState;
hashLookup.startSearch(hash2, searchState);
MultiStoreBytes entry = segmentState.tmpBytes;
@@ -1147,6 +1153,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
long sizeOfEverythingBeforeValue = entry.position();
alignment.alignPositionAddr(entry);
UpdateResult updateResult;
if (!isDeleted && prevValueSize == valueSize &&
metaValueInterop.startsWith(valueInterop, entry, value)) {
updateResult = UpdateResult.UNCHANGED;
@@ -1194,6 +1201,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
return updateResult;
}
// key is not found
checkPreincrementSize();
VBI valueInterop = getValueInterops.getValueInterop(copies);
MVBI metaValueInterop =
getValueInterops.getMetaValueInterop(copies, valueInterop, value);
@@ -1266,6 +1275,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
segmentState.identifier = identifier;
segmentState.timestamp = timestamp;
MultiMap hashLookup = hashLookup();
System.out.println("Before put size: " + hashLookup.size());
SearchState searchState = segmentState.searchState;
hashLookup.startSearch(hash2, searchState);
MultiStoreBytes entry = segmentState.tmpBytes;
@@ -1853,12 +1863,19 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
(position >= 0L && changes.nextSetBit(0L) != NOT_FOUND);
}
volatile Thread thread = null;
/**
* @param entryCallback call this to get an entry, this class will take care of the locking
* @return true if an entry was processed
*/
@Override
public boolean nextEntry(@NotNull final EntryCallback entryCallback, final int chronicleId) {
Thread t = Thread.currentThread();
if (thread == null)
thread = t;
else if (thread != t)
throw new AssertionError("Used by " + thread + " and " + t);
long position = this.position;
while (true) {
long oldPosition = position;
@@ -1884,8 +1901,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
entryCallback.onBeforeEntry();
final long segmentPos = position & posMask;
final Bytes entry =
segment.reuse(tmpBytes, segment.offsetFromPos(segmentPos));
long offset = segment.offsetFromPos(segmentPos);
final Bytes entry = segment.reuse2(tmpBytes, offset);
// it may not be successful if the buffer can not be re-sized so we will
// process it later, by NOT clearing the changes.clear(position)
src/main/java/net/openhft/chronicle/map/ShortShortMultiMap.java
+32 -15
View File
@@ -32,15 +32,15 @@ class ShortShortMultiMap implements MultiMap {
static final long MAX_CAPACITY = (1L << 16);
private static final long ENTRY_SIZE = 4L;
private static final int ENTRY_SIZE = 4;
private static final int ENTRY_SIZE_SHIFT = 2;
private static final int UNSET_KEY = 0;
private static final long MASK = 0xFFFFL;
private static final long HASH_INSTEAD_OF_UNSET_KEY = MASK;
private static final int UNSET_ENTRY = 0;
private final int capacity;
private final long capacityMask;
private final long capacityMask2;
private final int capacityMask;
private final int capacityMask2;
private final Bytes bytes;
private ATSDirectBitSet positions;
@@ -49,8 +49,8 @@ class ShortShortMultiMap implements MultiMap {
long capacity = multiMapCapacity(minCapacity);
assert isPowerOf2(capacity);
this.capacity = (int) capacity;
capacityMask = capacity - 1L;
capacityMask2 = capacityMask * ENTRY_SIZE;
capacityMask = (int) (capacity - 1L);
capacityMask2 = (int) (capacityMask * ENTRY_SIZE);
bytes = DirectStore.allocateLazy(capacity * ENTRY_SIZE).bytes();
positions = newPositions(capacity);
clear();
@@ -61,8 +61,8 @@ class ShortShortMultiMap implements MultiMap {
assert isPowerOf2(capacity);
assert (capacity / 2L) <= MAX_CAPACITY;
this.capacity = (int) capacity;
capacityMask = capacity - 1L;
capacityMask2 = capacityMask * ENTRY_SIZE;
capacityMask = (int) (capacity - 1L);
capacityMask2 = (int) (capacityMask * ENTRY_SIZE);
this.bytes = multiMapBytes;
positions = new ATSDirectBitSet(multiMapBitSetBytes);
}
@@ -90,7 +90,7 @@ class ShortShortMultiMap implements MultiMap {
return (key &= MASK) != UNSET_KEY ? key : HASH_INSTEAD_OF_UNSET_KEY;
}
private long step(long pos) {
private int step(int pos) {
return (pos + ENTRY_SIZE) & capacityMask2;
}
@@ -98,8 +98,8 @@ class ShortShortMultiMap implements MultiMap {
return (pos - ENTRY_SIZE) & capacityMask2;
}
private long pos(long key) {
return (key & capacityMask) << ENTRY_SIZE_SHIFT;
private int pos(long key) {
return ((int) key & capacityMask) << ENTRY_SIZE_SHIFT;
}
private static int entry(long key, long value) {
@@ -124,7 +124,7 @@ class ShortShortMultiMap implements MultiMap {
public void put(long key, long value) {
key = maskUnsetKey(key);
checkValueForPut(value);
for (long pos = pos(key); ; pos = step(pos)) {
for (int pos = pos(key); ; pos = step(pos)) {
int entry = bytes.readInt(pos);
if (entry == UNSET_ENTRY) {
bytes.writeInt(pos, entry(key, value));
@@ -146,7 +146,7 @@ class ShortShortMultiMap implements MultiMap {
checkValueForRemove(value);
long posToRemove;
int limit = capacity;
for (long pos = pos(key); ; pos = step(pos)) {
for (int pos = pos(key); ; pos = step(pos)) {
int entry = bytes.readInt(pos);
if (key(entry) == key && value(entry) == value) {
posToRemove = pos;
@@ -164,7 +164,7 @@ class ShortShortMultiMap implements MultiMap {
key = maskUnsetKey(key);
checkValueForRemove(oldValue);
checkValueForPut(newValue);
for (long pos = pos(key); ; pos = step(pos)) {
for (int pos = pos(key); ; pos = step(pos)) {
int entry = bytes.readInt(pos);
if (key(entry) == key && value(entry) == oldValue) {
positions.clear(oldValue);
@@ -176,7 +176,7 @@ class ShortShortMultiMap implements MultiMap {
}
private void removePos(long posToRemove) {
long posToShift = posToRemove;
int posToShift = (int) posToRemove;
while (true) {
posToShift = step(posToShift);
int entryToShift = bytes.readInt(posToShift);
@@ -207,9 +207,26 @@ class ShortShortMultiMap implements MultiMap {
searchStateToReuse.putAfterFailedSearch = false;
}
public long size() {
int pos = 0;
int size = 0;
for (int count = capacity; count > 0; count--) {
int entry = bytes.readInt(pos);
pos = step(pos);
if (entry != UNSET_ENTRY)
size++;
}
return size;
}
@Override
public long capacity() {
return capacity;
}
@Override
public long nextPos(SearchState searchState) {
long pos = searchState.searchPos;
int pos = (int) searchState.searchPos;
for(int count = capacity; count > 0; count--) {
int entry = bytes.readInt(pos);
if (entry == UNSET_ENTRY) {
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+39 -2
View File
@@ -1748,7 +1748,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
final Bytes segmentHeader;
final Bytes bytes;
final long entriesOffset;
private final int index;
private final int index, maxSize;
private final SingleThreadedDirectBitSet freeList;
long startWriteLock = 0;
private MultiMap hashLookup;
@@ -1761,9 +1761,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
this.segmentHeader = segmentHeader;
this.bytes = bytes;
this.index = index;
long start = bytes.startAddr();
hashLookup = createMultiMap(start);
maxSize = (int) (hashLookup.capacity() * 9 / 10);
start += CACHE_LINES.align(sizeOfMultiMap() + sizeOfMultiMapBitSet(), BYTES)
* multiMapsPerSegment();
final NativeBytes bsBytes = new NativeBytes(ms.objectSerializer(),
@@ -1982,6 +1982,14 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return entry;
}
final MultiStoreBytes reuse2(MultiStoreBytes entry, long offset) {
entry.setBytesOffset(bytes, offset);
entry.position(metaDataBytes);
if (bytes.readLong(offset) == 0)
Thread.yield();
return entry;
}
final long entrySize(long keySize, long valueSize) {
long sizeOfEverythingBeforeValue = sizeOfEverythingBeforeValue(keySize, valueSize);
return innerEntrySize(sizeOfEverythingBeforeValue, valueSize);
@@ -2205,7 +2213,10 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
if (prevValueSize == valueSize &&
metaValueInterop.startsWith(valueInterop, entry, value)) {
updateResult = UpdateResult.UNCHANGED;
} else {
checkPreincrementSize();
long valueAddr = entry.positionAddr();
long entryEndAddr = valueAddr + prevValueSize;
@@ -2284,6 +2295,21 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
}
}
protected void checkPreincrementSize() {
long size = size();
if (size > maxSize())
throw new IllegalStateException("Segment contains " + size + " with maxSize: " + maxSize() + ", capacity: " + capacity());
System.out.println("size: " + size + " of " + maxSize());
}
private long maxSize() {
return maxSize;
}
private long capacity() {
return entriesPerSegment;
}
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<RV, ? super VBI>>
RV put3(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState,
@@ -2421,6 +2447,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
}
if (doPutValue) {
if (entryIsDeleted)
checkPreincrementSize();
// putValue may relocate entry and change offset
putValue(pos, entry, valueSizePos, entryEndAddr,
entryIsDeleted, segmentState,
@@ -2490,6 +2519,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
clearMetaData(offset);
reuse(entry, offset);
long position = entry.position();
String addr = Long.toHexString(entry.address() + position);
keySizeMarshaller.writeSize(entry, keySize);
metaKeyInterop.write(keyInterop, entry, key);
@@ -2506,6 +2537,11 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
hashLookup.putAfterFailedSearch(segmentState.searchState, pos);
incrementSize();
long pos2 = entry.position();
entry.position(position);
System.out.println("write: " + addr + "\n"
+ entry.toHexString(32));
entry.position(pos2);
return valueSize;
}
@@ -2854,6 +2890,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
doPutValue = true;
}
if (doPutValue) {
checkPreincrementSize();
boolean entryIsDeleted = false; // couldn't replace deleted entry
putValue(pos, entry, valueSizePos, entryEndAddr, entryIsDeleted, segmentState,
metaValueInterop, valueInterop, newValue, newValueSize, searchedHashLookup,