diff --git a/pom.xml b/pom.xml index 89c69ece9..a59770bfb 100755 --- a/pom.xml +++ b/pom.xml @@ -44,7 +44,7 @@ net.openhft chronicle-bom - 1.10.19 + 1.10.28-SNAPSHOT pom import diff --git a/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java b/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java old mode 100644 new mode 100755 index dbe04622f..b36d8bf6c --- a/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java @@ -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; diff --git a/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java b/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java old mode 100644 new mode 100755 index 0fbc21617..eb62a1271 --- a/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java @@ -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; diff --git a/src/main/java/net/openhft/chronicle/map/MultiMap.java b/src/main/java/net/openhft/chronicle/map/MultiMap.java old mode 100644 new mode 100755 index bfa4e714e..3d1861ffa --- a/src/main/java/net/openhft/chronicle/map/MultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/MultiMap.java @@ -73,6 +73,10 @@ interface MultiMap { void forEach(EntryConsumer action); + long size(); + + long capacity(); + interface EntryConsumer { void accept(long key, long value); } diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java old mode 100644 new mode 100755 index 354b094b1..0a88a1061 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -468,6 +468,8 @@ final class ReplicatedChronicleMap 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= 0L; ) { @@ -932,6 +935,7 @@ final class ReplicatedChronicleMap= 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 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) { diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 70cf8a074..ccdb91b79 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -1748,7 +1748,7 @@ class VanillaChronicleMap, 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, 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, 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, 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, } } + 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; + } + , RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> RV put3(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState, @@ -2421,6 +2447,9 @@ class VanillaChronicleMap, } 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, 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, 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, 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,