diff --git a/pom.xml b/pom.xml index afe7a1c83..a8cf2bda9 100755 --- a/pom.xml +++ b/pom.xml @@ -15,7 +15,8 @@ ~ along with this program. If not, see . --> - + net.openhft @@ -23,7 +24,7 @@ 1.1.10 - + 4.0.0 chronicle-map 3.14.0-SNAPSHOT @@ -46,7 +47,7 @@ net.openhft chronicle-bom pom - 1.14.0-SNAPSHOT + 1.14.10-SNAPSHOT import diff --git a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java index 8f50db9cf..520301e75 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -71,112 +71,83 @@ public abstract class VanillaChronicleHash globalMutableStateLockTryAcquireOperation = + TryAcquireOperations.lock(); + static final + AcquisitionStrategy + globalMutableStateLockAcquisitionStrategy = + AcquisitionStrategies.spinLoopOrFail(2, TimeUnit.SECONDS); + private static final long GLOBAL_MUTABLE_STATE_LOCK_OFFSET = 0L; + private static final long GLOBAL_MUTABLE_STATE_VALUE_OFFSET = 8L; ///////////////////////////////////////////////// // If the hash was created in the first place, or read from disk public transient boolean createdOrInMemory; - ///////////////////////////////////////////////// // Key Data model public Class keyClass; public SizeMarshaller keySizeMarshaller; public SizedReader keyReader; public DataAccess keyDataAccess; - ///////////////////////////////////////////////// public boolean checksumEntries; - ///////////////////////////////////////////////// // Concurrency (number of segments), memory management and dependent fields public int actualSegments; public HashSplitting hashSplitting; - public long chunkSize; public int maxChunksPerEntry; public long actualChunksPerSegmentTier; - - ///////////////////////////////////////////////// - // Precomputed offsets and sizes for fast Context init - int segmentHeaderSize; - public int tierHashLookupValueBits; public int tierHashLookupKeyBits; public int tierHashLookupSlotSize; public long tierHashLookupCapacity; public long maxEntriesPerHashLookup; - long tierHashLookupInnerSize; public long tierHashLookupOuterSize; - public long tierFreeListInnerSize; public long tierFreeListOuterSize; - - long tierEntrySpaceInnerSize; public int tierEntrySpaceInnerOffset; - long tierEntrySpaceOuterSize; - public long tierSize; - + public long tiersInBulk; + public transient List tierBulkOffsets; + public transient long headerSize; + public transient long segmentHeadersOffset; + ///////////////////////////////////////////////// + // Miscellaneous fields + public transient CompactOffHeapLinearHashTable hashLookup; + public transient Identity identity; + protected int log2TiersInBulk; + ///////////////////////////////////////////////// + // Bytes Store (essentially, the base address) and serialization-dependent offsets + protected transient BytesStore bs; + ///////////////////////////////////////////////// + // Precomputed offsets and sizes for fast Context init + int segmentHeaderSize; + long tierHashLookupInnerSize; + long tierEntrySpaceInnerSize; + long tierEntrySpaceOuterSize; long maxExtraTiers; long tierBulkSizeInBytes; long tierBulkInnerOffsetToTiers; - public long tiersInBulk; - protected int log2TiersInBulk; - + transient long segmentsOffset; + ///////////////////////////////////////////////// + private String dataFileVersion; ///////////////////////////////////////////////// // Resources private transient File file; private transient RandomAccessFile raf; - private transient ChronicleHashResources resources; - private transient Cleaner cleaner; - - ///////////////////////////////////////////////// - // Bytes Store (essentially, the base address) and serialization-dependent offsets - protected transient BytesStore bs; - - public static class TierBulkData { - public final BytesStore bytesStore; - public final long offset; - - public TierBulkData(BytesStore bytesStore, long offset) { - this.bytesStore = bytesStore; - this.offset = offset; - } - - public TierBulkData(TierBulkData data, long offset) { - this.bytesStore = data.bytesStore; - this.offset = offset; - } - } - - public transient List tierBulkOffsets; - - public transient long headerSize; - public transient long segmentHeadersOffset; - transient long segmentsOffset; - - ///////////////////////////////////////////////// - // Miscellaneous fields - public transient CompactOffHeapLinearHashTable hashLookup; - - private transient VanillaGlobalMutableState globalMutableState; - - /** - * {@link ChronicleHashCloseOnExitHook} needs to use {@code VanillaChronicleHash}es as - * WeakHashMap keys, but with identity comparison, not Map's equals() and hashCode(). - */ - public class Identity { - public VanillaChronicleHash hash() { - return VanillaChronicleHash.this; - } - } - - public transient Identity identity; // --- End of instance fields --- - + private transient ChronicleHashResources resources; + private transient Cleaner cleaner; + private transient VanillaGlobalMutableState globalMutableState; public VanillaChronicleHash(ChronicleMapBuilder builder) { // Version @@ -238,6 +209,20 @@ public abstract class VanillaChronicleHash "keyClass").typeLiteral(); - keySizeMarshaller = wireIn.read(() -> "keySizeMarshaller").typedMarshallable(); - keyReader = wireIn.read(() -> "keyReader").typedMarshallable(); - keyDataAccess = wireIn.read(() -> "keyDataAccess").typedMarshallable(); + keySizeMarshaller = wireIn.read(() -> "keySizeMarshaller").object(SizeMarshaller.class); + keyReader = wireIn.read(() -> "keyReader").object(SizedReader.class); + keyDataAccess = wireIn.read(() -> "keyDataAccess").object(DataAccess.class); checksumEntries = wireIn.read(() -> "checksumEntries").bool(); @@ -297,14 +282,14 @@ public abstract class VanillaChronicleHash "dataFileVersion").text(dataFileVersion); wireOut.write(() -> "keyClass").typeLiteral(keyClass); - wireOut.write(() -> "keySizeMarshaller").typedMarshallable(keySizeMarshaller); - wireOut.write(() -> "keyReader").typedMarshallable(keyReader); - wireOut.write(() -> "keyDataAccess").typedMarshallable(keyDataAccess); + wireOut.write(() -> "keySizeMarshaller").object(keySizeMarshaller); + wireOut.write(() -> "keyReader").object(keyReader); + wireOut.write(() -> "keyDataAccess").object(keyDataAccess); wireOut.write(() -> "checksumEntries").bool(checksumEntries); wireOut.write(() -> "actualSegments").int32(actualSegments); - wireOut.write(() -> "hashSplitting").typedMarshallable(hashSplitting); + wireOut.write(() -> "hashSplitting").object(hashSplitting); wireOut.write(() -> "chunkSize").int64(chunkSize); wireOut.write(() -> "maxChunksPerEntry").int32(maxChunksPerEntry); @@ -424,20 +409,6 @@ public abstract class VanillaChronicleHash globalMutableStateLockTryAcquireOperation = - TryAcquireOperations.lock(); - static final - AcquisitionStrategy - globalMutableStateLockAcquisitionStrategy = - AcquisitionStrategies.spinLoopOrFail(2, TimeUnit.SECONDS); - - private static final long GLOBAL_MUTABLE_STATE_LOCK_OFFSET = 0L; - private static final long GLOBAL_MUTABLE_STATE_VALUE_OFFSET = 8L; - private long globalMutableStateAddress() { return bsAddress() + headerSize; } @@ -1045,4 +998,29 @@ public abstract class VanillaChronicleHash> allContexts() { return Collections.unmodifiableList(resources.contexts()); } + + public static class TierBulkData { + public final BytesStore bytesStore; + public final long offset; + + public TierBulkData(BytesStore bytesStore, long offset) { + this.bytesStore = bytesStore; + this.offset = offset; + } + + public TierBulkData(TierBulkData data, long offset) { + this.bytesStore = data.bytesStore; + this.offset = offset; + } + } + + /** + * {@link ChronicleHashCloseOnExitHook} needs to use {@code VanillaChronicleHash}es as + * WeakHashMap keys, but with identity comparison, not Map's equals() and hashCode(). + */ + public class Identity { + public VanillaChronicleHash hash() { + return VanillaChronicleHash.this; + } + } } diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/impl/SerializationBuilder.java b/src/main/java/net/openhft/chronicle/hash/serialization/impl/SerializationBuilder.java index 09a5a53c9..3e5d0780c 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/impl/SerializationBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/impl/SerializationBuilder.java @@ -20,6 +20,7 @@ package net.openhft.chronicle.hash.serialization.impl; import net.openhft.chronicle.bytes.Byteable; import net.openhft.chronicle.bytes.BytesMarshallable; import net.openhft.chronicle.core.OS; +import net.openhft.chronicle.core.annotation.NotNull; import net.openhft.chronicle.core.util.ReadResolvable; import net.openhft.chronicle.hash.serialization.*; import net.openhft.chronicle.values.ValueModel; @@ -47,7 +48,7 @@ public final class SerializationBuilder implements Cloneable { configureByDefault(tClass); sizeIsStaticallyKnown = constantSizeMarshaller(); } - + private static boolean concreteClass(Class c) { return !c.isInterface() && !Modifier.isAbstract(c.getModifiers()); } @@ -93,12 +94,7 @@ public final class SerializationBuilder implements Cloneable { if (concreteClass(tClass) && Byteable.class.isAssignableFrom(tClass)) { reader(new ByteableSizedReader<>((Class) tClass)); dataAccess((DataAccess) new ByteableDataAccess<>((Class) tClass)); - try { - long byteableSize = ((Byteable) OS.memory().allocateInstance(tClass)).maxSize(); - sizeMarshaller(constant(byteableSize)); - } catch (InstantiationException e) { - throw new IllegalStateException(e); - } + sizeMarshaller(constant(allocateByteable(tClass).maxSize())); } else if (tClass == CharSequence.class) { reader((SizedReader) CharSequenceSizedReader.INSTANCE); dataAccess((DataAccess) new CharSequenceUtf8DataAccess()); @@ -142,6 +138,15 @@ public final class SerializationBuilder implements Cloneable { } } + @NotNull + private Byteable allocateByteable(Class tClass) { + try { + return (Byteable) OS.memory().allocateInstance(tClass); + } catch (InstantiationException e) { + throw new AssertionError(e); + } + } + public void reader(SizedReader reader) { checkNonMarshallableEnum(reader.getClass()); this.reader = reader; diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 77cf0a1d7..fa99e0e87 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -152,8 +152,6 @@ public final class ChronicleMapBuilder implements private static final double UNDEFINED_DOUBLE_CONFIG = Double.NaN; private static final ConcurrentHashMap fileLockingControl = new ConcurrentHashMap<>(128); - private static int MAX_BOOTSTRAPPING_HEADER_SIZE = (int) MemoryUnit.KILOBYTES.toBytes(16); - private static final Logger chronicleMapLogger = LoggerFactory.getLogger(ChronicleMap.class); private static final ChronicleHashCorruption.Listener defaultChronicleMapCorruptionListener = corruption -> { @@ -163,8 +161,7 @@ public final class ChronicleMapBuilder implements chronicleMapLogger.error(corruption.message()); } }; - - private String name; + private static int MAX_BOOTSTRAPPING_HEADER_SIZE = (int) MemoryUnit.KILOBYTES.toBytes(16); SerializationBuilder keyBuilder; SerializationBuilder valueBuilder; K averageKey; @@ -178,7 +175,6 @@ public final class ChronicleMapBuilder implements long cleanupTimeout = 1; TimeUnit cleanupTimeoutUnit = TimeUnit.MINUTES; boolean cleanupRemovedEntries = true; - ////////////////////////////// // Configuration fields DefaultValueProvider defaultValueProvider = DefaultSpi.defaultValueProvider(); @@ -186,6 +182,7 @@ public final class ChronicleMapBuilder implements MapMethods methods = DefaultSpi.mapMethods(); MapEntryOperations entryOperations = mapEntryOperations(); MapRemoteOperations remoteOperations = mapRemoteOperations(); + private String name; // not final because of cloning private ChronicleMapBuilderPrivateAPI privateAPI = new ChronicleMapBuilderPrivateAPI<>(this); @@ -421,6 +418,52 @@ public final class ChronicleMapBuilder implements writeFully(fileChannel, SIZE_WORD_OFFSET, headerBuffer); } + /** + * @return ByteBuffer, in [position, limit) range the self bootstrapping header is read + */ + private static ByteBuffer readSelfBootstrappingHeader( + File file, RandomAccessFile raf, int headerSize, boolean recover, + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) throws IOException { + if (raf.length() < headerSize + SELF_BOOTSTRAPPING_HEADER_OFFSET) { + throw throwRecoveryOrReturnIOException(file, + "The file is shorter than the header size: " + headerSize + + ", file size: " + raf.length(), recover); + } + FileChannel fileChannel = raf.getChannel(); + ByteBuffer headerBuffer = ByteBuffer.allocate( + SELF_BOOTSTRAPPING_HEADER_OFFSET + headerSize); + headerBuffer.order(LITTLE_ENDIAN); + readFully(fileChannel, 0, headerBuffer); + if (headerBuffer.remaining() > 0) { + throw throwRecoveryOrReturnIOException(file, "Unable to read the header fully, " + + headerBuffer.remaining() + " is remaining to read, likely the file was " + + "truncated", recover); + } + int sizeWord = headerBuffer.getInt(SIZE_WORD_OFFSET); + if (!SizePrefixedBlob.isReady(sizeWord)) { + if (recover) { + report(corruptionListener, corruption, -1, () -> + format("file={}: size-prefixed blob readiness bit is set to NOT_COMPLETE", + file) + ); + // the bit will be overwritten to READY in the end of recovery procedure, so nothing + // to fix right here + } else { + throw new IOException("file=" + file + ": sizeWord is not ready: " + sizeWord); + } + } + headerBuffer.position(SELF_BOOTSTRAPPING_HEADER_OFFSET); + return headerBuffer; + } + + private static boolean checkSumSelfBootstrappingHeader( + ByteBuffer headerBuffer, int headerSize) { + long checkSum = headerChecksum(headerBuffer, headerSize); + long storedChecksum = headerBuffer.getLong(HEADER_OFFSET); + return storedChecksum == checkSum; + } + @Override public ChronicleMapBuilder clone() { try { @@ -1666,52 +1709,6 @@ public final class ChronicleMapBuilder implements } } - /** - * @return ByteBuffer, in [position, limit) range the self bootstrapping header is read - */ - private static ByteBuffer readSelfBootstrappingHeader( - File file, RandomAccessFile raf, int headerSize, boolean recover, - ChronicleHashCorruption.Listener corruptionListener, - ChronicleHashCorruptionImpl corruption) throws IOException { - if (raf.length() < headerSize + SELF_BOOTSTRAPPING_HEADER_OFFSET) { - throw throwRecoveryOrReturnIOException(file, - "The file is shorter than the header size: " + headerSize + - ", file size: " + raf.length(), recover); - } - FileChannel fileChannel = raf.getChannel(); - ByteBuffer headerBuffer = ByteBuffer.allocate( - SELF_BOOTSTRAPPING_HEADER_OFFSET + headerSize); - headerBuffer.order(LITTLE_ENDIAN); - readFully(fileChannel, 0, headerBuffer); - if (headerBuffer.remaining() > 0) { - throw throwRecoveryOrReturnIOException(file, "Unable to read the header fully, " + - headerBuffer.remaining() + " is remaining to read, likely the file was " + - "truncated", recover); - } - int sizeWord = headerBuffer.getInt(SIZE_WORD_OFFSET); - if (!SizePrefixedBlob.isReady(sizeWord)) { - if (recover) { - report(corruptionListener, corruption, -1, () -> - format("file={}: size-prefixed blob readiness bit is set to NOT_COMPLETE", - file) - ); - // the bit will be overwritten to READY in the end of recovery procedure, so nothing - // to fix right here - } else { - throw new IOException("file=" + file+ ": sizeWord is not ready: " + sizeWord); - } - } - headerBuffer.position(SELF_BOOTSTRAPPING_HEADER_OFFSET); - return headerBuffer; - } - - private static boolean checkSumSelfBootstrappingHeader( - ByteBuffer headerBuffer, int headerSize) { - long checkSum = headerChecksum(headerBuffer, headerSize); - long storedChecksum = headerBuffer.getLong(HEADER_OFFSET); - return storedChecksum == checkSum; - } - private VanillaChronicleMap createWithNewFile( VanillaChronicleMap map, File file, RandomAccessFile raf, ChronicleHashResources resources, ByteBuffer headerBuffer, int headerSize) @@ -1752,6 +1749,7 @@ public final class ChronicleMapBuilder implements Bytes headerBytes = Bytes.wrapForRead(headerBuffer); headerBytes.readPosition(headerBuffer.position()); headerBytes.readLimit(headerBuffer.limit()); + System.out.println(headerBytes); Wire wire = new TextWire(headerBytes); VanillaChronicleMap map = wire.getValueIn().typedMarshallable(); map.initBeforeMapping(file, raf, headerBuffer.limit(), recover); diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 079316be8..99f8b0198 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -56,20 +56,29 @@ public class VanillaChronicleMap ExternalMapQueryContext> implements AbstractChronicleMap { - ///////////////////////////////////////////////// - // Value Data model - Class valueClass; public SizeMarshaller valueSizeMarshaller; public SizedReader valueReader; public DataAccess valueDataAccess; - public boolean constantlySizedEntry; - ///////////////////////////////////////////////// // Memory management and dependent fields public int alignment; public int worstAlignment; - + public transient boolean couldNotDetermineAlignmentBeforeAllocation; + /** @see net.openhft.chronicle.set.SetFromMap */ + public transient ChronicleSet chronicleSet; + public transient MapEntryOperations entryOperations; + public transient MapMethods methods; + public transient DefaultValueProvider defaultValueProvider; + ///////////////////////////////////////////////// + // Value Data model + Class valueClass; + ///////////////////////////////////////////////// + // Behavior + transient boolean putReturnsNull; + transient boolean removeReturnsNull; + transient Set> entrySet; + transient ThreadLocal cxt; ///////////////////////////////////////////////// private transient String name; /** @@ -78,25 +87,7 @@ public class VanillaChronicleMap * initOwnTransients(). */ private transient String identityString; - - public transient boolean couldNotDetermineAlignmentBeforeAllocation; - - ///////////////////////////////////////////////// - // Behavior - transient boolean putReturnsNull; - transient boolean removeReturnsNull; - - transient Set> entrySet; - - /** @see net.openhft.chronicle.set.SetFromMap */ - public transient ChronicleSet chronicleSet; - - public transient MapEntryOperations entryOperations; - public transient MapMethods methods; private transient boolean defaultEntryOperationsAndMethods; - public transient DefaultValueProvider defaultValueProvider; - - transient ThreadLocal cxt; public VanillaChronicleMap(ChronicleMapBuilder builder) throws IOException { super(builder); @@ -116,14 +107,18 @@ public class VanillaChronicleMap initTransients(); } + public static long alignAddr(long addr, long alignment) { + return (addr + alignment - 1) & ~(alignment - 1L); + } + @Override protected void readMarshallableFields(@NotNull WireIn wireIn) { super.readMarshallableFields(wireIn); valueClass = wireIn.read(() -> "valueClass").typeLiteral(); - valueSizeMarshaller = wireIn.read(() -> "valueSizeMarshaller").typedMarshallable(); - valueReader = wireIn.read(() -> "valueReader").typedMarshallable(); - valueDataAccess = wireIn.read(() -> "valueDataAccess").typedMarshallable(); + valueSizeMarshaller = wireIn.read(() -> "valueSizeMarshaller").object(SizeMarshaller.class); + valueReader = wireIn.read(() -> "valueReader").object(SizedReader.class); + valueDataAccess = wireIn.read(() -> "valueDataAccess").object(DataAccess.class); constantlySizedEntry = wireIn.read(() -> "constantlySizedEntry").bool(); @@ -136,9 +131,9 @@ public class VanillaChronicleMap super.writeMarshallable(wireOut); wireOut.write(() -> "valueClass").typeLiteral(valueClass); - wireOut.write(() -> "valueSizeMarshaller").typedMarshallable(valueSizeMarshaller); - wireOut.write(() -> "valueReader").typedMarshallable(valueReader); - wireOut.write(() -> "valueDataAccess").typedMarshallable(valueDataAccess); + wireOut.write(() -> "valueSizeMarshaller").object(valueSizeMarshaller); + wireOut.write(() -> "valueReader").object(valueReader); + wireOut.write(() -> "valueDataAccess").object(valueDataAccess); wireOut.write(() -> "constantlySizedEntry").bool(constantlySizedEntry); @@ -287,7 +282,7 @@ public class VanillaChronicleMap ", identityHashCode=" + System.identityHashCode(this) + "}"; } - + @Override public void clear() { forEachEntry(c -> c.context().remove(c)); @@ -306,10 +301,6 @@ public class VanillaChronicleMap entry.readSkip(skip); } - public static long alignAddr(long addr, long alignment) { - return (addr + alignment - 1) & ~(alignment - 1L); - } - final ChainingInterface q() { ThreadLocal cxt = this.cxt; if (cxt == null) diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/data/ZeroBytesStore.java b/src/main/java/net/openhft/chronicle/map/impl/stage/data/ZeroBytesStore.java index 454da16ce..0e81772fd 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/data/ZeroBytesStore.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/data/ZeroBytesStore.java @@ -231,4 +231,6 @@ public enum ZeroBytesStore implements BytesStore { public void nativeWrite(long l, long l1, long l2) { throw new UnsupportedOperationException(); } + + } \ No newline at end of file diff --git a/src/test/java/net/openhft/chronicle/map/Issue125Test.java b/src/test/java/net/openhft/chronicle/map/Issue125Test.java new file mode 100644 index 000000000..be118218e --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/Issue125Test.java @@ -0,0 +1,48 @@ +/* + * Copyright 2016 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.chronicle.core.OS; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; + +/** + * Created by peter on 04/06/17. + */ +public class Issue125Test { + @Test + public void test() throws IOException { + final File cacheRoot = new File(OS.TARGET + "/test.cm3"); + ChronicleMapBuilder shaToNodeBuilder = + ChronicleMapBuilder.of(byte[].class, byte[].class) +// .name("bytes-to-bytes") + .entries(1000000). + averageKeySize(20). + averageValueSize(30); + + ChronicleMap shaToNode = + shaToNodeBuilder.createPersistedTo(cacheRoot); + + shaToNode.put("1".getBytes(), "2".getBytes()); + + shaToNodeBuilder.createPersistedTo(cacheRoot); + + } + +}