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

Add a fix for enums being written as marshallables.

Browse Source
This commit is contained in:
Peter Lawrey
2017-06-04 14:01:48 +01:00
parent 52d8a0e6e3
commit ed8361b5cf
7 changed files with 227 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
+86 -108
View File
@@ -71,112 +71,83 @@ public abstract class VanillaChronicleHash<K,
public static final long RESERVED_GLOBAL_MUTABLE_STATE_BYTES = 1024;
// --- Start of instance fields ---
/////////////////////////////////////////////////
private String dataFileVersion;
/**
* Global mutable state lock doesn't yet need read-write levels and waits;
* Used the same locking strategy as in segment locks
* (VanillaReadWriteUpdateWithWaitsLockingStrategy) in order to simplify Chronicle Map
* specification (having only one kind of locks to specify and implement).
*/
static final LockingStrategy globalMutableStateLockingStrategy =
VanillaReadWriteUpdateWithWaitsLockingStrategy.instance();
static final TryAcquireOperation<LockingStrategy> globalMutableStateLockTryAcquireOperation =
TryAcquireOperations.lock();
static final
AcquisitionStrategy<LockingStrategy, RuntimeException>
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<K> keyClass;
public SizeMarshaller keySizeMarshaller;
public SizedReader<K> keyReader;
public DataAccess<K> 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<TierBulkData> 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<TierBulkData> 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<K, ?> builder) {
// Version
@@ -238,6 +209,20 @@ public abstract class VanillaChronicleHash<K,
checksumEntries = privateAPI.checksumEntries();
}
public static IOException throwRecoveryOrReturnIOException(
File file, String message, boolean recover) {
message = "file=" + file + " " + message;
if (recover) {
throw new ChronicleHashRecoveryFailedException(message);
} else {
return new IOException(message);
}
}
private static long roundUpMapHeaderSize(long headerSize) {
return CACHE_LINES.align(headerSize, BYTES);
}
@Override
public void readMarshallable(@NotNull WireIn wire) {
readMarshallableFields(wire);
@@ -253,9 +238,9 @@ public abstract class VanillaChronicleHash<K,
createdOrInMemory = false;
keyClass = wireIn.read(() -> "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<K,
wireOut.write(() -> "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<K,
}
}
public static IOException throwRecoveryOrReturnIOException(
File file, String message, boolean recover) {
message = "file=" + file + " " + message;
if (recover) {
throw new ChronicleHashRecoveryFailedException(message);
} else {
return new IOException(message);
}
}
private static long roundUpMapHeaderSize(long headerSize) {
return CACHE_LINES.align(headerSize, BYTES);
}
public final void createInMemoryStoreAndSegments(ChronicleHashResources resources)
throws IOException {
this.resources = resources;
@@ -739,24 +710,6 @@ public abstract class VanillaChronicleHash<K,
return actualSegments;
}
/**
* Global mutable state lock doesn't yet need read-write levels and waits;
* Used the same locking strategy as in segment locks
* (VanillaReadWriteUpdateWithWaitsLockingStrategy) in order to simplify Chronicle Map
* specification (having only one kind of locks to specify and implement).
*/
static final LockingStrategy globalMutableStateLockingStrategy =
VanillaReadWriteUpdateWithWaitsLockingStrategy.instance();
static final TryAcquireOperation<LockingStrategy> globalMutableStateLockTryAcquireOperation =
TryAcquireOperations.lock();
static final
AcquisitionStrategy<LockingStrategy, RuntimeException>
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<K,
public List<WeakReference<ContextHolder>> 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;
}
}
}
src/main/java/net/openhft/chronicle/hash/serialization/impl/SerializationBuilder.java
+12 -7
View File
@@ -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<T> 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<T> implements Cloneable {
if (concreteClass(tClass) && Byteable.class.isAssignableFrom(tClass)) {
reader(new ByteableSizedReader<>((Class) tClass));
dataAccess((DataAccess<T>) 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<T>) CharSequenceSizedReader.INSTANCE);
dataAccess((DataAccess<T>) new CharSequenceUtf8DataAccess());
@@ -142,6 +138,15 @@ public final class SerializationBuilder<T> implements Cloneable {
}
}
@NotNull
private Byteable allocateByteable(Class<T> tClass) {
try {
return (Byteable) OS.memory().allocateInstance(tClass);
} catch (InstantiationException e) {
throw new AssertionError(e);
}
}
public void reader(SizedReader<T> reader) {
checkNonMarshallableEnum(reader.getClass());
this.reader = reader;
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+49 -51
View File
@@ -152,8 +152,6 @@ public final class ChronicleMapBuilder<K, V> implements
private static final double UNDEFINED_DOUBLE_CONFIG = Double.NaN;
private static final ConcurrentHashMap<File, Void> 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<K, V> implements
chronicleMapLogger.error(corruption.message());
}
};
private String name;
private static int MAX_BOOTSTRAPPING_HEADER_SIZE = (int) MemoryUnit.KILOBYTES.toBytes(16);
SerializationBuilder<K> keyBuilder;
SerializationBuilder<V> valueBuilder;
K averageKey;
@@ -178,7 +175,6 @@ public final class ChronicleMapBuilder<K, V> implements
long cleanupTimeout = 1;
TimeUnit cleanupTimeoutUnit = TimeUnit.MINUTES;
boolean cleanupRemovedEntries = true;
//////////////////////////////
// Configuration fields
DefaultValueProvider<K, V> defaultValueProvider = DefaultSpi.defaultValueProvider();
@@ -186,6 +182,7 @@ public final class ChronicleMapBuilder<K, V> implements
MapMethods<K, V, ?> methods = DefaultSpi.mapMethods();
MapEntryOperations<K, V, ?> entryOperations = mapEntryOperations();
MapRemoteOperations<K, V, ?> remoteOperations = mapRemoteOperations();
private String name;
// not final because of cloning
private ChronicleMapBuilderPrivateAPI<K, V> privateAPI =
new ChronicleMapBuilderPrivateAPI<>(this);
@@ -421,6 +418,52 @@ public final class ChronicleMapBuilder<K, V> 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<K, V> clone() {
try {
@@ -1666,52 +1709,6 @@ public final class ChronicleMapBuilder<K, V> 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<K, V, ?> createWithNewFile(
VanillaChronicleMap<K, V, ?> map, File file, RandomAccessFile raf,
ChronicleHashResources resources, ByteBuffer headerBuffer, int headerSize)
@@ -1752,6 +1749,7 @@ public final class ChronicleMapBuilder<K, V> implements
Bytes<ByteBuffer> headerBytes = Bytes.wrapForRead(headerBuffer);
headerBytes.readPosition(headerBuffer.position());
headerBytes.readLimit(headerBuffer.limit());
System.out.println(headerBytes);
Wire wire = new TextWire(headerBytes);
VanillaChronicleMap<K, V, ?> map = wire.getValueIn().typedMarshallable();
map.initBeforeMapping(file, raf, headerBuffer.limit(), recover);
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+26 -35
View File
@@ -56,20 +56,29 @@ public class VanillaChronicleMap<K, V, R>
ExternalMapQueryContext<K, V, ?>>
implements AbstractChronicleMap<K, V> {
/////////////////////////////////////////////////
// Value Data model
Class<V> valueClass;
public SizeMarshaller valueSizeMarshaller;
public SizedReader<V> valueReader;
public DataAccess<V> 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<K> chronicleSet;
public transient MapEntryOperations<K, V, R> entryOperations;
public transient MapMethods<K, V, R> methods;
public transient DefaultValueProvider<K, V> defaultValueProvider;
/////////////////////////////////////////////////
// Value Data model
Class<V> valueClass;
/////////////////////////////////////////////////
// Behavior
transient boolean putReturnsNull;
transient boolean removeReturnsNull;
transient Set<Entry<K, V>> entrySet;
transient ThreadLocal<ContextHolder> cxt;
/////////////////////////////////////////////////
private transient String name;
/**
@@ -78,25 +87,7 @@ public class VanillaChronicleMap<K, V, R>
* initOwnTransients().
*/
private transient String identityString;
public transient boolean couldNotDetermineAlignmentBeforeAllocation;
/////////////////////////////////////////////////
// Behavior
transient boolean putReturnsNull;
transient boolean removeReturnsNull;
transient Set<Entry<K, V>> entrySet;
/** @see net.openhft.chronicle.set.SetFromMap */
public transient ChronicleSet<K> chronicleSet;
public transient MapEntryOperations<K, V, R> entryOperations;
public transient MapMethods<K, V, R> methods;
private transient boolean defaultEntryOperationsAndMethods;
public transient DefaultValueProvider<K, V> defaultValueProvider;
transient ThreadLocal<ContextHolder> cxt;
public VanillaChronicleMap(ChronicleMapBuilder<K, V> builder) throws IOException {
super(builder);
@@ -116,14 +107,18 @@ public class VanillaChronicleMap<K, V, R>
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<K, V, R>
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<K, V, R>
", identityHashCode=" + System.identityHashCode(this) +
"}";
}
@Override
public void clear() {
forEachEntry(c -> c.context().remove(c));
@@ -306,10 +301,6 @@ public class VanillaChronicleMap<K, V, R>
entry.readSkip(skip);
}
public static long alignAddr(long addr, long alignment) {
return (addr + alignment - 1) & ~(alignment - 1L);
}
final ChainingInterface q() {
ThreadLocal<ContextHolder> cxt = this.cxt;
if (cxt == null)
src/main/java/net/openhft/chronicle/map/impl/stage/data/ZeroBytesStore.java
+2
View File
@@ -231,4 +231,6 @@ public enum ZeroBytesStore implements BytesStore<ZeroBytesStore, Void> {
public void nativeWrite(long l, long l1, long l2) {
throw new UnsupportedOperationException();
}
}