diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshaller.java index 5c988298d..abf375ec1 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshaller.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshaller.java @@ -26,6 +26,8 @@ public interface SizeMarshaller extends Serializable { int sizeEncodingSize(long size); + long minEncodableSize(); + int minSizeEncodingSize(); int maxSizeEncodingSize(); diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshallers.java b/src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshallers.java index 490a34a58..758c603e7 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshallers.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshallers.java @@ -36,6 +36,11 @@ public final class SizeMarshallers { return stopBitLength(size); } + @Override + public long minEncodableSize() { + return 0L; + } + @Override public int minSizeEncodingSize() { return 1; @@ -75,6 +80,11 @@ public final class SizeMarshallers { return 0; } + @Override + public long minEncodableSize() { + return size; + } + @Override public int minSizeEncodingSize() { return 0; diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java index b6350cca5..db3682f36 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java @@ -71,6 +71,11 @@ public abstract class ByteableMarshaller return 0; } + @Override + public long minEncodableSize() { + return size; + } + @Override public int minSizeEncodingSize() { return 0; diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/internal/DoubleMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/internal/DoubleMarshaller.java index 6df3b6c79..b54ca23db 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/internal/DoubleMarshaller.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/internal/DoubleMarshaller.java @@ -39,6 +39,12 @@ public enum DoubleMarshaller implements BytesInterop, BytesReader implements Cloneable, size += averageSizeEncodingSize(valueBuilder, valueSize); Alignment alignment = valueAlignment(); int worstAlignment; - if (alignment != Alignment.NO_ALIGNMENT) { - if (constantlySizedKeys() && valueBuilder.constantSizeEncodingSizeMarshaller()) { - long constantSizeBeforeAlignment = round(size); - if (constantlySizedValues()) { - // see specialEntrySpaceOffset() - long totalDataSize = constantSizeBeforeAlignment + constantValueSize(); - worstAlignment = (int) (alignment.alignAddr(totalDataSize) - totalDataSize); - } else { - determineAlignment: - if (actualChunkSize > 0) { - worstAlignment = worstAlignmentAssumingChunkSize(constantSizeBeforeAlignment, - actualChunkSize); - } else { - int chunkSize = 8; - worstAlignment = worstAlignmentAssumingChunkSize( - constantSizeBeforeAlignment, chunkSize); - if (size + worstAlignment + valueSize >= - maxDefaultChunksPerAverageEntry(replicated) * chunkSize) { - break determineAlignment; - } - chunkSize = 4; - worstAlignment = worstAlignmentAssumingChunkSize( - constantSizeBeforeAlignment, chunkSize); - } - } + if (alignment != Alignment.NO_ALIGNMENT && + constantlySizedKeys() && valueBuilder.constantSizeEncodingSizeMarshaller()) { + long constantSizeBeforeAlignment = round(size); + if (constantlySizedValues()) { + // see specialEntrySpaceOffset() + long totalDataSize = constantSizeBeforeAlignment + constantValueSize(); + worstAlignment = (int) (alignment.alignAddr(totalDataSize) - totalDataSize); } else { - // assume worst case, we always lose most possible bytes for alignment - worstAlignment = alignment.alignment() - 1; + determineAlignment: + if (actualChunkSize > 0) { + worstAlignment = worstAlignmentAssumingChunkSize(constantSizeBeforeAlignment, + actualChunkSize); + } else { + int chunkSize = 8; + worstAlignment = worstAlignmentAssumingChunkSize( + constantSizeBeforeAlignment, chunkSize); + if (size + worstAlignment + valueSize >= + maxDefaultChunksPerAverageEntry(replicated) * chunkSize) { + break determineAlignment; + } + chunkSize = 4; + worstAlignment = worstAlignmentAssumingChunkSize( + constantSizeBeforeAlignment, chunkSize); + } } } else { - worstAlignment = 0; + // assume worst case, we always lose most possible bytes for alignment + worstAlignment = alignment.alignment() - 1; } size += worstAlignment; size += valueSize; @@ -481,7 +478,7 @@ public final class ChronicleMapBuilder implements Cloneable, * always the same, we should _misalign_ the first chunk. */ int specialEntrySpaceOffset(boolean replicated) { - if (!constantlySizedEntries() || valueAlignment() == Alignment.NO_ALIGNMENT) + if (!constantlySizedEntries()) return 0; return (int) (constantValueSize() % valueAlignment().alignment()); } @@ -510,8 +507,6 @@ public final class ChronicleMapBuilder implements Cloneable, private int worstAlignmentAssumingChunkSize( long constantSizeBeforeAlignment, int chunkSize) { - if (valueAlignment() == Alignment.NO_ALIGNMENT) - return 0; Alignment valueAlignment = valueAlignment(); long firstAlignment = valueAlignment.alignAddr(constantSizeBeforeAlignment) - constantSizeBeforeAlignment; diff --git a/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java b/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java index 5e3349f54..ca137ab23 100644 --- a/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java +++ b/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java @@ -80,8 +80,8 @@ enum MultiMapFactory { throw new IllegalArgumentException("Capacity " + capacity + " not supported"); } - public static long sizeOfBitSetInBytes(long minCapacity) { - return BYTES.convert(LONGS.align(multiMapCapacity(minCapacity), BITS), BITS); + public static long sizeOfBitSetInBytes(long actualChunksPerSegment) { + return LONGS.align(BYTES.alignAndConvert(actualChunksPerSegment, BITS), BYTES); } public static ATSDirectBitSet newPositions(long capacity) { diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 86756d749..25cfaccaa 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -683,19 +683,8 @@ final class ReplicatedChronicleMap Integer.MAX_VALUE) - throw new IllegalStateException("ENTRY WRITE_BUFFER_SIZE TOO LARGE : Replicated " + - "ChronicleMap's" + - " are restricted to an " + - "entry size of " + Integer.MAX_VALUE + ", " + - "your entry size=" + result); - - return result; + long sizeOfEverythingBeforeValue(long keySize, long valueSize) { + return super.sizeOfEverythingBeforeValue(keySize, valueSize) + ADDITIONAL_ENTRY_BYTES; } @Override @@ -735,7 +724,9 @@ final class ReplicatedChronicleMap, assert booleanResult || expectedValue == null; } + static final boolean checkSegmentMultiMapsAndBitSetsConsistencyOnBootstrap = Boolean.getBoolean( + "chronicleMap.checkSegmentMultiMapsAndBitSetsConsistencyOnBootstrap"); + // these methods should be package local, not public or private. class Segment implements SharedSegment { /* @@ -1421,7 +1423,6 @@ class VanillaChronicleMap, */ static final long LOCK_OFFSET = 0L; // 64-bit static final long SIZE_OFFSET = LOCK_OFFSET + 8L; // 32-bit - ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); final NativeBytes segmentHeader; final NativeBytes bytes; final long entriesOffset; @@ -1448,13 +1449,15 @@ class VanillaChronicleMap, * multiMapsPerSegment(); final NativeBytes bsBytes = new NativeBytes(ms.objectSerializer(), start, - start + LONGS.align(BYTES.alignAndConvert(actualChunksPerSegment, BITS), BYTES), + start + MultiMapFactory.sizeOfBitSetInBytes(actualChunksPerSegment), null); freeList = new SingleThreadedDirectBitSet(bsBytes); start += numberOfBitSets() * sizeOfSegmentFreeListBitSets(); start += specialEntrySpaceOffset; entriesOffset = start - bytes.startAddr(); assert bytes.capacity() >= entriesOffset + actualChunksPerSegment * chunkSize; + if (checkSegmentMultiMapsAndBitSetsConsistencyOnBootstrap) + checkMultiMapsAndBitSetsConsistency(); } final MultiMap hashLookup() { @@ -1473,6 +1476,30 @@ class VanillaChronicleMap, return multiMapFactory.create(multiMapBytes, sizeOfMultiMapBitSetBytes); } + void checkMultiMapsAndBitSetsConsistency() { + MultiMap hashLookup = hashLookup(); + final DirectBitSet positions = hashLookup.getPositions(); + class EntryChecker implements MultiMap.EntryConsumer { + long size = 0; + @Override + public void accept(long key, long value) { + if (positions.isSet(value)) + size++; + if (freeList.isClear(value)) + throw new IllegalStateException("Position " + value + " is present in " + + "multiMap but available in the free chunk list"); + } + } + EntryChecker entryChecker = new EntryChecker(); + hashLookup.forEach(entryChecker); + if (size() != entryChecker.size || entryChecker.size != positions.cardinality()) { + throw new IllegalStateException("Segment inconsistent: " + + "size by Segment counter: " + size() + + ", size by multiMap records present in bit set: " + entryChecker.size + + ", size by multiMap bit set cardinality: " + positions.cardinality()); + } + } + public final int getIndex() { return index; } @@ -1614,10 +1641,12 @@ class VanillaChronicleMap, return entry; } - long entrySize(long keySize, long valueSize) { - long sizeOfEverythingBeforeValue = metaDataBytes + - keySizeMarshaller.sizeEncodingSize(keySize) + keySize + - valueSizeMarshaller.sizeEncodingSize(valueSize); + final long entrySize(long keySize, long valueSize) { + long sizeOfEverythingBeforeValue = sizeOfEverythingBeforeValue(keySize, valueSize); + return innerEntrySize(sizeOfEverythingBeforeValue, valueSize); + } + + private long innerEntrySize(long sizeOfEverythingBeforeValue, long valueSize) { if (constantlySizedEntry) { return alignment.alignAddr(sizeOfEverythingBeforeValue + valueSize); } else if (couldNotDetermineAlignmentBeforeAllocation) { @@ -1627,6 +1656,12 @@ class VanillaChronicleMap, } } + long sizeOfEverythingBeforeValue(long keySize, long valueSize) { + return metaDataBytes + + keySizeMarshaller.sizeEncodingSize(keySize) + keySize + + valueSizeMarshaller.sizeEncodingSize(valueSize); + } + final int inChunks(long sizeInBytes) { if (sizeInBytes <= chunkSize) return 1; @@ -1862,7 +1897,9 @@ class VanillaChronicleMap, long valueSize = metaValueInterop.size(valueInterop, value); long valueSizePos = entry.position(); - long prevValueSize = readValueSize(entry); + long prevValueSize = valueSizeMarshaller.readSize(entry); + long sizeOfEverythingBeforeValue = entry.position(); + alignment.alignPositionAddr(entry); long valueAddr = entry.positionAddr(); long entryEndAddr = valueAddr + prevValueSize; @@ -1878,7 +1915,8 @@ class VanillaChronicleMap, // putValue may relocate entry and change offset putValue(pos, offset, entry, valueSizePos, entryEndAddr, segmentState, - metaValueInterop, valueInterop, value, valueSize, searchedHashLookup); + metaValueInterop, valueInterop, value, valueSize, searchedHashLookup, + sizeOfEverythingBeforeValue); // put callbacks onPutMaybeRemote(segmentState.pos, remote); @@ -1912,7 +1950,8 @@ class VanillaChronicleMap, MultiStoreBytes entry, boolean writeDefaultInitialReplicationValues) { long valueSize = metaElemWriter.size(elemWriter, elem); long entrySize = entrySize(keySize, valueSize); - long pos = alloc(inChunks(entrySize)); + int allocatedChunks = inChunks(entrySize); + long pos = alloc(allocatedChunks); segmentState.pos = pos; long offset = offsetFromPos(pos); clearMetaData(offset); @@ -1928,12 +1967,22 @@ class VanillaChronicleMap, alignment.alignPositionAddr(entry); metaElemWriter.write(elemWriter, entry, elem); + freeExtraAllocatedChunks(pos, allocatedChunks, entry); + hashLookup.putAfterFailedSearch(segmentState.searchState, pos); incrementSize(); return valueSize; } + final void freeExtraAllocatedChunks(long pos, int allocatedChunks, Bytes entry) { + int actuallyUsedChunks; + if (!constantlySizedEntry && couldNotDetermineAlignmentBeforeAllocation && // fast path + (actuallyUsedChunks = inChunks(entry.position())) < allocatedChunks) { + free(pos + actuallyUsedChunks, allocatedChunks - actuallyUsedChunks); + } + } + void manageReplicationBytes(Bytes entry, boolean writeDefaultInitialReplicationValues, boolean remove) { // do nothing @@ -2217,7 +2266,9 @@ class VanillaChronicleMap, ReadValue readValue, InstanceOrBytesToInstance toValue, long pos, long offset, MultiStoreBytes entry, MultiMap searchedHashLookup) { long valueSizePos = entry.position(); - long valueSize = readValueSize(entry); + long valueSize = valueSizeMarshaller.readSize(entry); + long sizeOfEverythingBeforeValue = entry.position(); + alignment.alignPositionAddr(entry); long entryEndAddr = entry.positionAddr() + valueSize; RV prevValue; @@ -2241,7 +2292,8 @@ class VanillaChronicleMap, copies, valueInterop, newValue); long newValueSize = metaValueInterop.size(valueInterop, newValue); putValue(pos, offset, entry, valueSizePos, entryEndAddr, segmentState, - metaValueInterop, valueInterop, newValue, newValueSize, searchedHashLookup); + metaValueInterop, valueInterop, newValue, newValueSize, searchedHashLookup, + sizeOfEverythingBeforeValue); // put callbacks onPut(this, segmentState.pos); @@ -2275,7 +2327,7 @@ class VanillaChronicleMap, SegmentState segmentState, MetaBytesWriter metaElemWriter, EW elemWriter, E newElem, long newElemSize, - MultiMap searchedHashLookup) { + MultiMap searchedHashLookup, long sizeOfEverythingBeforeValue) { long entryStartAddr = entry.address(); long valueSizeAddr = entryStartAddr + valueSizePos; long newValueAddr = alignment.alignAddr( @@ -2297,7 +2349,9 @@ class VanillaChronicleMap, // RELOCATION free(pos, oldSizeInChunks); onRelocation(this, pos); - pos = alloc(newSizeInChunks); + int allocatedChunks = + inChunks(innerEntrySize(sizeOfEverythingBeforeValue, newElemSize)); + pos = alloc(allocatedChunks); // putValue() is called from put() and replace() // after successful search by key searchedHashLookup.replacePrevPos(segmentState.searchState, pos); @@ -2309,6 +2363,15 @@ class VanillaChronicleMap, NativeBytes.UNSAFE.copyMemory(entryStartAddr, newEntryStartAddr, valueSizeAddr - entryStartAddr); entry = reuse(entry, offset); + entry.position(valueSizePos); + valueSizeMarshaller.writeSize(entry, newElemSize); + alignment.alignPositionAddr(entry); + metaElemWriter.write(elemWriter, entry, newElem); + + freeExtraAllocatedChunks(pos, allocatedChunks, entry); + + segmentState.pos = pos; + return offset; // END OF RELOCATION } else if (newSizeInChunks < oldSizeInChunks) { // Freeing extra chunks @@ -2748,7 +2811,9 @@ class VanillaChronicleMap, entry.skip(keySize); segment.manageReplicationBytes(entry, true, false); long valueSizePos = entry.position(); - long valueSize = map.readValueSize(entry); + long valueSize = map.valueSizeMarshaller.readSize(entry); + long sizeOfEverythingBeforeValue = entry.position(); + map.alignment.alignPositionAddr(entry); long entryEndAddr = entry.positionAddr() + valueSize; VI valueInterop = map.valueInteropProvider.get(copies, map.originalValueInterop); V value = value(); @@ -2756,7 +2821,8 @@ class VanillaChronicleMap, .get(copies, map.originalMetaValueInterop, valueInterop, value); long newValueSize = metaValueInterop.size(valueInterop, value); segment.putValue(pos, offset, entry, valueSizePos, entryEndAddr, segmentState, - metaValueInterop, valueInterop, value, newValueSize, segment.hashLookup()); + metaValueInterop, valueInterop, value, newValueSize, segment.hashLookup(), + sizeOfEverythingBeforeValue); map.onPut(segment, segmentState.pos); } putOnClose = true; diff --git a/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java b/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java index 877757afa..feca01b7f 100644 --- a/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java +++ b/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java @@ -64,6 +64,11 @@ enum DummyValueMarshaller return 0; } + @Override + public long minEncodableSize() { + return 0L; + } + @Override public int minSizeEncodingSize() { return 0; diff --git a/src/main/java/net/openhft/xstreem/converters/AbstractChronicleMapConverter.java b/src/main/java/net/openhft/xstreem/converters/AbstractChronicleMapConverter.java index ee18b3c44..e84844863 100644 --- a/src/main/java/net/openhft/xstreem/converters/AbstractChronicleMapConverter.java +++ b/src/main/java/net/openhft/xstreem/converters/AbstractChronicleMapConverter.java @@ -77,9 +77,12 @@ public class AbstractChronicleMapConverter implements Converter { @Override public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext context) { - reader.moveDown(); + // empty map + if ("[\"\"]".equals(reader.getValue())) + return null; if (!"cmap".equals(reader.getNodeName())) - throw new ConversionException("root node should be cmap, was " + reader.getNodeName()); + throw new ConversionException("should be under 'cmap' node"); + reader.moveDown(); while (reader.hasMoreChildren()) { reader.moveDown(); diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java index 591756425..62ed81f0b 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java @@ -53,8 +53,12 @@ public class TCPSocketReplicationTest { ChronicleMapBuilder map1Builder = newTcpSocketShmBuilder(Integer.class, CharSequence.class, (byte) 1, port, new InetSocketAddress("localhost", port + 1)); - map1 = map1Builder.averageValueSize(10).create(); - map2 = newTcpSocketShmIntString((byte) 2, port + 1); + map1 = map1Builder.entries(Builder.SIZE).averageValueSize(10).create(); + + ChronicleMapBuilder map2Builder = + newTcpSocketShmBuilder(Integer.class, CharSequence.class, + (byte) 2, port + 1); + map2 = map2Builder.entries(Builder.SIZE).averageValueSize(10).create(); s_port += 2; }