diff --git a/pom.xml b/pom.xml index a838bb414..68acc6ecd 100755 --- a/pom.xml +++ b/pom.xml @@ -52,7 +52,7 @@ net.openhft lang - 6.4.7-SNAPSHOT + 6.4.8-SNAPSHOT diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index eddc5782a..be93dadf3 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -16,35 +16,33 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.map.serialization.MetaBytesInterop; +import net.openhft.chronicle.map.serialization.MetaBytesWriter; import net.openhft.lang.Maths; import net.openhft.lang.io.*; import net.openhft.lang.io.serialization.*; import net.openhft.lang.io.serialization.impl.*; -import net.openhft.lang.model.Byteable; -import net.openhft.lang.model.DataValueClasses; import org.jetbrains.annotations.NotNull; -import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.*; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import static net.openhft.chronicle.map.Objects.builderEquals; -import static net.openhft.chronicle.map.Objects.hash; public class ChronicleMapBuilder implements Cloneable { public static final short UDP_REPLICATION_MODIFICATION_ITERATOR_ID = 128; private static final Logger LOG = LoggerFactory.getLogger(ChronicleMapBuilder.class.getName()); - final Class keyClass; - final Class valueClass; + SerializationBuilder keyBuilder; + SerializationBuilder valueBuilder; boolean transactional = false; Replicator firstReplicator; - Map, Replicator> replicators = - new HashMap, Replicator>(); + Map, Replicator> replicators = new HashMap, Replicator>(); boolean forceReplicatedImpl = false; // used when configuring the number of segments. private int minSegments = -1; @@ -66,60 +64,18 @@ public class ChronicleMapBuilder implements Cloneable { private TimeProvider timeProvider = TimeProvider.SYSTEM; private BytesMarshallerFactory bytesMarshallerFactory; private ObjectSerializer objectSerializer; - @NotNull - private BytesMarshaller keyMarshaller; - @NotNull - private BytesMarshaller valueMarshaller; - @NotNull - private ObjectFactory valueFactory; private MapEventListener> eventListener = MapEventListeners.nop(); ChronicleMapBuilder(Class keyClass, Class valueClass) { - this.keyClass = keyClass; - this.valueClass = valueClass; - keyMarshaller = chooseDefaultMarshaller(keyClass); - valueMarshaller = chooseDefaultMarshaller(valueClass); - valueFactory = marshallerUseFactory(valueClass) ? - new AllocateInstanceObjectFactory(valueClass.isInterface() ? - DataValueClasses.directClassFor(valueClass) : - valueClass) : - NullObjectFactory.INSTANCE; + keyBuilder = new SerializationBuilder(keyClass, SerializationBuilder.Role.KEY); + valueBuilder = new SerializationBuilder(valueClass, SerializationBuilder.Role.VALUE); } public static ChronicleMapBuilder of(Class keyClass, Class valueClass) { return new ChronicleMapBuilder(keyClass, valueClass); } - private static boolean marshallerUseFactory(Class c) { - return Byteable.class.isAssignableFrom(c) || - BytesMarshallable.class.isAssignableFrom(c) || - Externalizable.class.isAssignableFrom(c); - } - - @SuppressWarnings("unchecked") - private static BytesMarshaller chooseDefaultMarshaller(@NotNull Class tClass) { - Class classForMarshaller = marshallerUseFactory(tClass) && tClass.isInterface() ? - DataValueClasses.directClassFor(tClass) : tClass; - if (Byteable.class.isAssignableFrom(tClass)) - return new ByteableMarshaller(classForMarshaller); - if (BytesMarshallable.class.isAssignableFrom(tClass)) - return new BytesMarshallableMarshaller(classForMarshaller); - if (Externalizable.class.isAssignableFrom(tClass)) - return new ExternalizableMarshaller(classForMarshaller); - if (tClass == CharSequence.class) - return (BytesMarshaller) CharSequenceMarshaller.INSTANCE; - if (tClass == String.class) - return (BytesMarshaller) StringMarshaller.INSTANCE; - if (tClass == Integer.class) - return (BytesMarshaller) IntegerMarshaller.INSTANCE; - if (tClass == Long.class) - return (BytesMarshaller) LongMarshaller.INSTANCE; - if (tClass == Double.class) - return (BytesMarshaller) DoubleMarshaller.INSTANCE; - return SerializableMarshaller.INSTANCE; - } - private static long roundUpMapHeaderSize(long headerSize) { long roundUp = (headerSize + 127L) & ~127L; if (roundUp - headerSize < 64) @@ -129,12 +85,12 @@ public class ChronicleMapBuilder implements Cloneable { @Override public ChronicleMapBuilder clone() { - try { @SuppressWarnings("unchecked") - final ChronicleMapBuilder result = (ChronicleMapBuilder) super.clone(); + final ChronicleMapBuilder result = (ChronicleMapBuilder) super.clone(); + result.keyBuilder = keyBuilder.clone(); + result.valueBuilder = valueBuilder.clone(); return result; - } catch (CloneNotSupportedException e) { throw new AssertionError(e); } @@ -393,12 +349,8 @@ public class ChronicleMapBuilder implements Cloneable { ", largeSegments=" + largeSegments() + ", timeProvider=" + timeProvider() + ", bytesMarshallerfactory=" + bytesMarshallerFactory() + - - ", keyClass=" + keyClass + - ", valueClass=" + valueClass + - ", keyMarshaller=" + keyMarshaller + - ", valueMarshaller=" + valueMarshaller + - ", valueFactory=" + valueFactory + + ", keyBuilder=" + keyBuilder + + ", valueBuilder=" + valueBuilder + ", eventListener=" + eventListener + '}'; } @@ -481,76 +433,19 @@ public class ChronicleMapBuilder implements Cloneable { return this; } - @NotNull - public BytesMarshaller keyMarshaller() { - return keyMarshaller; - } - - public ChronicleMapBuilder keyMarshaller( - @NotNull BytesMarshaller keyMarshaller) { - this.keyMarshaller = keyMarshaller; + public ChronicleMapBuilder keyMarshaller(@NotNull BytesMarshaller keyMarshaller) { + keyBuilder.marshaller(keyMarshaller, null); return this; } - @NotNull - public BytesMarshaller valueMarshaller() { - return valueMarshaller; - } - public ChronicleMapBuilder valueMarshallerAndFactory( @NotNull BytesMarshaller valueMarshaller, @NotNull ObjectFactory valueFactory) { - this.valueMarshaller = valueMarshaller; - this.valueFactory = valueFactory; + valueBuilder.marshaller(valueMarshaller, valueFactory); return this; } - @NotNull - public ObjectFactory valueFactory() { - return valueFactory; - } - - @SuppressWarnings("unchecked") - public ChronicleMapBuilder valueFactory( - @NotNull ObjectFactory valueFactory) { - if (!marshallerUseFactory(valueClass)) { - throw new IllegalStateException("Default marshaller for " + valueClass + - " value don't use object factory"); - } else if (valueMarshaller instanceof ByteableMarshaller) { - if (valueFactory instanceof AllocateInstanceObjectFactory) { - valueMarshaller = new ByteableMarshaller( - ((AllocateInstanceObjectFactory) valueFactory).allocatedClass()); - } else { - valueMarshaller = new ByteableMarshallerWithCustomFactory( - ((ByteableMarshaller) valueMarshaller).tClass, valueFactory); - } - } else if (valueMarshaller instanceof BytesMarshallableMarshaller) { - if (valueFactory instanceof AllocateInstanceObjectFactory) { - valueMarshaller = new BytesMarshallableMarshaller( - ((AllocateInstanceObjectFactory) valueFactory).allocatedClass()); - } else { - valueMarshaller = new BytesMarshallableMarshallerWithCustomFactory( - ((BytesMarshallableMarshaller) valueMarshaller).marshaledClass(), - valueFactory - ); - } - } else if (valueMarshaller instanceof ExternalizableMarshaller) { - if (valueFactory instanceof AllocateInstanceObjectFactory) { - valueMarshaller = new ExternalizableMarshaller( - ((AllocateInstanceObjectFactory) valueFactory).allocatedClass()); - } else { - valueMarshaller = new ExternalizableMarshallerWithCustomFactory( - ((ExternalizableMarshaller) valueMarshaller).marshaledClass(), - valueFactory - ); - } - } else { - // valueMarshaller is custom, it is user's responsibility to use the same factory inside - // marshaller and standalone - throw new IllegalStateException( - "Change the value factory simultaneously with marshaller " + - "using valueMarshallerAndFactory() method"); - } - this.valueFactory = valueFactory; + public ChronicleMapBuilder valueFactory(@NotNull ObjectFactory valueFactory) { + valueBuilder.factory(valueFactory); return this; } @@ -570,7 +465,8 @@ public class ChronicleMapBuilder implements Cloneable { FileInputStream fis = new FileInputStream(file); ObjectInputStream ois = new ObjectInputStream(fis); try { - VanillaChronicleMap map = (VanillaChronicleMap) ois.readObject(); + VanillaChronicleMap map = + (VanillaChronicleMap) ois.readObject(); map.headerSize = roundUpMapHeaderSize(fis.getChannel().position()); map.createMappedStoreAndSegments(file); return establishReplication(map); @@ -593,7 +489,7 @@ public class ChronicleMapBuilder implements Cloneable { if (!file.exists()) throw new FileNotFoundException("Unable to create " + file); - VanillaChronicleMap map = newMap(); + VanillaChronicleMap map = newMap(); FileOutputStream fos = new FileOutputStream(file); ObjectOutputStream oos = new ObjectOutputStream(fos); @@ -610,20 +506,29 @@ public class ChronicleMapBuilder implements Cloneable { } public ChronicleMap create() throws IOException { - VanillaChronicleMap map = newMap(); + VanillaChronicleMap map = newMap(); BytesStore bytesStore = new DirectStore(objectSerializer(), map.sizeInBytes(), true); map.createMappedStoreAndSegments(bytesStore); return establishReplication(map); } - private VanillaChronicleMap newMap() throws IOException { + private VanillaChronicleMap newMap() throws IOException { + preMapConstruction(); if (firstReplicator == null && !forceReplicatedImpl) { - return new VanillaChronicleMap(this); + return new VanillaChronicleMap, + V, Object, MetaBytesWriter>(this); } else { - return new ReplicatedChronicleMap(this); + return new ReplicatedChronicleMap, + V, Object, MetaBytesWriter>(this); } } + private void preMapConstruction() { + int maxSize = entrySize() * figureBufferAllocationFactor(); + keyBuilder.maxSize(maxSize); + valueBuilder.maxSize(maxSize); + } + private ChronicleMap establishReplication(ChronicleMap map) throws IOException { if (map instanceof ReplicatedChronicleMap) { @@ -646,312 +551,11 @@ public class ChronicleMapBuilder implements Cloneable { return map; } - private static enum CharSequenceMarshaller implements BytesMarshaller { - INSTANCE; - - @Override - public void write(Bytes bytes, CharSequence s) { - bytes.writeUTFΔ(s); - } - - @Nullable - @Override - public CharSequence read(Bytes bytes) { - return bytes.readUTFΔ(); - } - - @Nullable - @Override - public CharSequence read(Bytes bytes, @Nullable CharSequence s) { - if (s instanceof StringBuilder) { - if (bytes.readUTFΔ((StringBuilder) s)) - return s; - return null; - } - return bytes.readUTFΔ(); - } - } - - private static enum StringMarshaller implements BytesMarshaller { - INSTANCE; - - @Override - public void write(Bytes bytes, String s) { - bytes.writeUTFΔ(s); - } - - @Nullable - @Override - public String read(Bytes bytes) { - return bytes.readUTFΔ(); - } - - @Nullable - @Override - public String read(Bytes bytes, @Nullable String s) { - return bytes.readUTFΔ(); - } - } - - private static enum IntegerMarshaller implements BytesMarshaller { - INSTANCE; - - @Override - public void write(Bytes bytes, Integer v) { - bytes.writeInt(v); - } - - @Nullable - @Override - public Integer read(Bytes bytes) { - return bytes.readInt(); - } - - @Nullable - @Override - public Integer read(Bytes bytes, @Nullable Integer v) { - return bytes.readInt(); - } - } - - private static enum LongMarshaller implements BytesMarshaller { - INSTANCE; - - @Override - public void write(Bytes bytes, Long v) { - bytes.writeLong(v); - } - - @Nullable - @Override - public Long read(Bytes bytes) { - return bytes.readLong(); - } - - @Nullable - @Override - public Long read(Bytes bytes, @Nullable Long v) { - return bytes.readLong(); - } - } - - private static enum DoubleMarshaller implements BytesMarshaller { - INSTANCE; - - @Override - public void write(Bytes bytes, Double v) { - bytes.writeDouble(v); - } - - @Nullable - @Override - public Double read(Bytes bytes) { - return bytes.readDouble(); - } - - @Nullable - @Override - public Double read(Bytes bytes, @Nullable Double v) { - return bytes.readDouble(); - } - } - - - private static enum SerializableMarshaller implements BytesMarshaller { - INSTANCE; - - @Override - public void write(Bytes bytes, Object obj) { - bytes.writeObject(obj); - } - - @Nullable - @Override - public Object read(Bytes bytes) { - return bytes.readObject(); - } - - @Nullable - @Override - public Object read(Bytes bytes, @Nullable Object obj) { - return bytes.readInstance(null, obj); - } - } - - private static class ByteableMarshaller implements BytesMarshaller { - private static final long serialVersionUID = 0L; - - @NotNull - final Class tClass; - - ByteableMarshaller(@NotNull Class tClass) { - this.tClass = tClass; - } - - @Override - public void write(Bytes bytes, T t) { - bytes.write(t.bytes(), t.offset(), t.maxSize()); - } - - @Nullable - @Override - public T read(Bytes bytes) { - return read(bytes, null); - } - - @Nullable - @Override - public T read(Bytes bytes, @Nullable T t) { - try { - if (t == null) - t = getInstance(); - t.bytes(bytes, bytes.position()); - bytes.skip(t.maxSize()); - return t; - } catch (Exception e) { - throw new IllegalStateException(e); - } - } - - @SuppressWarnings("unchecked") - @NotNull - T getInstance() throws Exception { - return (T) NativeBytes.UNSAFE.allocateInstance(tClass); - } - - @Override - public boolean equals(Object obj) { - return obj != null && obj.getClass() == getClass() && - ((ByteableMarshaller) obj).tClass == tClass; - } - - @Override - public int hashCode() { - return tClass.hashCode(); - } - - @Override - public String toString() { - return getClass().getSimpleName() + "{tClass=" + tClass + "}"; - } - } - - private static class ByteableMarshallerWithCustomFactory - extends ByteableMarshaller { - private static final long serialVersionUID = 0L; - - @NotNull - private final ObjectFactory factory; - - ByteableMarshallerWithCustomFactory(@NotNull Class tClass, - @NotNull ObjectFactory factory) { - super(tClass); - this.factory = factory; - } - - @NotNull - @Override - T getInstance() throws Exception { - return factory.create(); - } - - @Override - public boolean equals(Object obj) { - if (obj == null || obj.getClass() != getClass()) - return false; - ByteableMarshallerWithCustomFactory that = (ByteableMarshallerWithCustomFactory) obj; - return that.tClass == tClass && that.factory.equals(this.factory); - } - - @Override - public int hashCode() { - return hash(tClass, factory); - } - - @Override - public String toString() { - return getClass().getSimpleName() + "{tClass=" + tClass + ",factory=" + factory + "}"; - } - } - - private static class BytesMarshallableMarshallerWithCustomFactory - extends BytesMarshallableMarshaller { - private static final long serialVersionUID = 0L; - - @NotNull - private final ObjectFactory factory; - - BytesMarshallableMarshallerWithCustomFactory(@NotNull Class tClass, - @NotNull ObjectFactory factory) { - super(tClass); - this.factory = factory; - } - - @NotNull - @Override - protected T getInstance() throws Exception { - return factory.create(); - } - - @Override - public boolean equals(Object obj) { - if (obj == null || obj.getClass() != getClass()) - return false; - BytesMarshallableMarshallerWithCustomFactory that = - (BytesMarshallableMarshallerWithCustomFactory) obj; - return that.marshaledClass() == marshaledClass() && that.factory.equals(this.factory); - } - - @Override - public int hashCode() { - return hash(marshaledClass(), factory); - } - - @Override - public String toString() { - return getClass().getSimpleName() + "{marshaledClass=" + marshaledClass() + - ",factory=" + factory + "}"; - } - } - - private static class ExternalizableMarshallerWithCustomFactory - extends ExternalizableMarshaller { - private static final long serialVersionUID = 0L; - - @NotNull - private final ObjectFactory factory; - - ExternalizableMarshallerWithCustomFactory(@NotNull Class tClass, - @NotNull ObjectFactory factory) { - super(tClass); - this.factory = factory; - } - - @NotNull - @Override - protected T getInstance() throws Exception { - return factory.create(); - } - - @Override - public boolean equals(Object obj) { - if (obj == null || obj.getClass() != getClass()) - return false; - ExternalizableMarshallerWithCustomFactory that = - (ExternalizableMarshallerWithCustomFactory) obj; - return that.marshaledClass() == marshaledClass() && that.factory.equals(this.factory); - } - - @Override - public int hashCode() { - return hash(marshaledClass(), factory); - } - - @Override - public String toString() { - return getClass().getSimpleName() + "{marshaledClass=" + marshaledClass() + - ",factory=" + factory + "}"; - } + private int figureBufferAllocationFactor() { + // if expected map size is about 1000, seems rather wasteful to allocate + // key and value serialization buffers each x64 of expected entry size.. + return (int) Math.min(Math.max(2L, entries() >> 10), + VanillaChronicleMap.MAX_ENTRY_OVERSIZE_FACTOR); } } diff --git a/src/main/java/net/openhft/chronicle/map/MapEventListener.java b/src/main/java/net/openhft/chronicle/map/MapEventListener.java index 4822551e6..68b4e88a2 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEventListener.java +++ b/src/main/java/net/openhft/chronicle/map/MapEventListener.java @@ -40,7 +40,7 @@ public abstract class MapEventListener> * @param usingValue value provided to reuse, could be null. * @return null if null should be returned, or a value to put in the map and return. */ - public V onGetMissing(M map, Bytes keyBytes, K key, V usingValue) { + public V onGetMissing(M map, K key, V usingValue) { return null; } diff --git a/src/main/java/net/openhft/chronicle/map/MapEventListeners.java b/src/main/java/net/openhft/chronicle/map/MapEventListeners.java index 51df6b807..e07915ba6 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEventListeners.java +++ b/src/main/java/net/openhft/chronicle/map/MapEventListeners.java @@ -36,14 +36,9 @@ public final class MapEventListeners { public final Logger LOGGER = LoggerFactory.getLogger(getClass()); @Override - public Object onGetMissing(ChronicleMap map, Bytes keyBytes, + public Object onGetMissing(ChronicleMap map, Object key, Object usingValue) { - StringBuilder sb = new StringBuilder(); - sb.append(map.file()).append(" missed "); - - keyBytes.toString(sb, 0, 0, keyBytes.limit()); - - LOGGER.info(sb.toString()); + LOGGER.info("{} missed {}", map.file(), key); return null; } @@ -62,13 +57,14 @@ public final class MapEventListeners { sb.append(" | "); } Bytes slice = entry.slice(metaDataBytes, entry.limit() - metaDataBytes); - long keyLength = slice.readStopBit(); - slice.toString(sb, slice.position(), 0L, slice.position() + keyLength); - slice.position(slice.position() + keyLength); - long valueLength = slice.readStopBit(); - slice.alignPositionAddr(4); + VanillaChronicleMap vanillaMap = (VanillaChronicleMap) map; + long keySize = vanillaMap.keySizeMarshaller.readSize(slice); + slice.toString(sb, slice.position(), 0L, slice.position() + keySize); + slice.position(slice.position() + keySize); + long valueSize = vanillaMap.valueSizeMarshaller.readSize(slice); + vanillaMap.alignment.alignPositionAddr(slice); sb.append(" = "); - slice.toString(sb, slice.position(), 0L, slice.position() + valueLength); + slice.toString(sb, slice.position(), 0L, slice.position() + valueSize); LOGGER.info(sb.toString()); } @@ -89,7 +85,7 @@ public final class MapEventListeners { public final Logger LOGGER = LoggerFactory.getLogger(getClass()); @Override - public Object onGetMissing(ChronicleMap map, Bytes keyBytes, + public Object onGetMissing(ChronicleMap map, Object key, Object usingValue) { LOGGER.info("{} missed {}", map.file(), key); return null; diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index d4d000ae4..5c7df7df2 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -20,10 +20,13 @@ package net.openhft.chronicle.map; import net.openhft.lang.Maths; import net.openhft.lang.collection.ATSDirectBitSet; import net.openhft.lang.io.Bytes; -import net.openhft.lang.io.DirectBytes; import net.openhft.lang.io.MultiStoreBytes; import net.openhft.lang.io.NativeBytes; +import net.openhft.chronicle.map.serialization.BytesReader; +import net.openhft.chronicle.map.serialization.MetaBytesInterop; +import net.openhft.chronicle.map.serialization.MetaBytesWriter; import net.openhft.lang.model.Byteable; +import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,7 +37,7 @@ import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.atomic.AtomicReferenceArray; -import static net.openhft.chronicle.map.VanillaChronicleMap.Hasher.hash; +import static net.openhft.chronicle.map.serialization.Hasher.hash; import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND; /** @@ -72,7 +75,9 @@ import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND; * @param the entries key type * @param the entries value type */ -class ReplicatedChronicleMap extends VanillaChronicleMap +class ReplicatedChronicleMap, + V, VW, MVW extends MetaBytesWriter> + extends VanillaChronicleMap implements ChronicleMap, Replica, Replica.EntryExternalizable, Replica.EntryResolver, Closeable { // for file, jdbc and UDP replication @@ -193,7 +198,6 @@ class ReplicatedChronicleMap extends VanillaChronicleMap * @return the previous value * @see #put(Object, Object) */ - V put(K key, V value, byte identifier, long timeStamp) { assert identifier > 0; return put0(key, value, true, identifier, timeStamp); @@ -216,12 +220,16 @@ class ReplicatedChronicleMap extends VanillaChronicleMap final byte identifier, long timeStamp) { checkKey(key); checkValue(value); - Bytes keyBytes = getKeyAsBytes(key); - long hash = Hasher.hash(keyBytes); - int segmentNum = hasher.getSegment(hash); - int segmentHash = hasher.segmentHash(hash); - return segment(segmentNum).put(keyBytes, key, value, segmentHash, replaceIfPresent, - identifier, timeStamp); + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyWriter = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyWriter = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); + long hash = metaKeyWriter.hash(keyWriter, key); + int segmentNum = getSegment(hash); + int segmentHash = segmentHash(hash); + return segment(segmentNum).put(copies, metaKeyWriter, keyWriter, key, value, segmentHash, + replaceIfPresent, identifier, timeStamp); } /** @@ -235,12 +243,16 @@ class ReplicatedChronicleMap extends VanillaChronicleMap @Override V lookupUsing(K key, V value, boolean create) { checkKey(key); - Bytes keyBytes = getKeyAsBytes(key); - long hash = Hasher.hash(keyBytes); - int segmentNum = hasher.getSegment(hash); - int segmentHash = hasher.segmentHash(hash); - return segment(segmentNum).acquire(keyBytes, key, value, segmentHash, create, - timeProvider.currentTimeMillis()); + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyWriter = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyWriter = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); + long hash = metaKeyWriter.hash(keyWriter, key); + int segmentNum = getSegment(hash); + int segmentHash = segmentHash(hash); + return segment(segmentNum).acquire(copies, metaKeyWriter, keyWriter, key, value, + segmentHash, create, timeProvider.currentTimeMillis()); } @Override @@ -310,27 +322,36 @@ class ReplicatedChronicleMap extends VanillaChronicleMap localIdentifier, timeProvider.currentTimeMillis()) != null; } - private V removeIfValueIs(final Object key, final V expectedValue, + private V removeIfValueIs(final Object k, final V expectedValue, final byte identifier, final long timestamp) { - checkKey(key); - Bytes keyBytes = getKeyAsBytes((K) key); - long hash = Hasher.hash(keyBytes); - int segmentNum = hasher.getSegment(hash); - int segmentHash = hasher.segmentHash(hash); - return segment(segmentNum).remove(keyBytes, (K) key, expectedValue, segmentHash, - timestamp, identifier); + checkKey(k); + K key = (K) k; + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyWriter = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyWriter = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); + long hash = metaKeyWriter.hash(keyWriter, key); + int segmentNum = getSegment(hash); + int segmentHash = segmentHash(hash); + return segment(segmentNum).remove(copies, metaKeyWriter, keyWriter, key, expectedValue, + segmentHash, timestamp, identifier); } @Override V replaceIfValueIs(@net.openhft.lang.model.constraints.NotNull final K key, final V existingValue, final V newValue) { checkKey(key); checkValue(newValue); - Bytes keyBytes = getKeyAsBytes(key); - long hash = Hasher.hash(keyBytes); - int segmentNum = hasher.getSegment(hash); - int segmentHash = hasher.segmentHash(hash); - return segment(segmentNum).replace(keyBytes, key, existingValue, newValue, segmentHash, - timeProvider.currentTimeMillis()); + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyWriter = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyWriter = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); + long hash = metaKeyWriter.hash(keyWriter, key); + int segmentNum = getSegment(hash); + int segmentHash = segmentHash(hash); + return segment(segmentNum).replace(copies, metaKeyWriter, keyWriter, key, existingValue, + newValue, segmentHash, timeProvider.currentTimeMillis()); } /** @@ -343,9 +364,9 @@ class ReplicatedChronicleMap extends VanillaChronicleMap final long initialLimit = entry.limit(); - final long keyLen = entry.readStopBit(); + final long keySize = keySizeMarshaller.readSize(entry); final long keyPosition = entry.position(); - entry.skip(keyLen); + entry.skip(keySize); final long keyLimit = entry.position(); final long timeStamp = entry.readLong(); @@ -356,18 +377,18 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } final boolean isDeleted = entry.readBoolean(); - long valueLen; + long valueSize; if (!isDeleted) { - valueLen = entry.readStopBit(); - assert valueLen > 0; + valueSize = valueSizeMarshaller.readSize(entry); + assert valueSize > 0; } else { - valueLen = 0; + valueSize = 0; } final long valuePosition = entry.position(); - destination.writeStopBit(keyLen); - destination.writeStopBit(valueLen); + keySizeMarshaller.writeSize(destination, keySize); + valueSizeMarshaller.writeSize(destination, valueSize); destination.writeStopBit(timeStamp); // we store the isDeleted flag in the identifier @@ -401,7 +422,7 @@ class ReplicatedChronicleMap extends VanillaChronicleMap alignment.alignPositionAddr(entry); // writes the value - entry.limit(entry.position() + valueLen); + entry.limit(entry.position() + valueSize); destination.write(entry); if (debugEnabled) { @@ -416,8 +437,8 @@ class ReplicatedChronicleMap extends VanillaChronicleMap @Override public void readExternalEntry(@NotNull Bytes source) { - final long keyLen = source.readStopBit(); - final long valueLen = source.readStopBit(); + final long keySize = keySizeMarshaller.readSize(source); + final long valueSize = valueSizeMarshaller.readSize(source); final long timeStamp = source.readStopBit(); final byte id = source.readByte(); final byte remoteIdentifier; @@ -439,13 +460,13 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } final long keyPosition = source.position(); - final long keyLimit = keyPosition + keyLen; + final long keyLimit = keyPosition + keySize; source.limit(keyLimit); - long hash = Hasher.hash(source); + long hash = hash(source); - int segmentNum = hasher.getSegment(hash); - int segmentHash = hasher.segmentHash(hash); + int segmentNum = getSegment(hash); + int segmentHash = segmentHash(hash); boolean debugEnabled = LOG.isDebugEnabled(); @@ -471,7 +492,7 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } final long valuePosition = keyLimit; - final long valueLimit = valuePosition + valueLen; + final long valueLimit = valuePosition + valueSize; segment(segmentNum).remotePut(source, segmentHash, remoteIdentifier, timeStamp, valuePosition, valueLimit); setLastModificationTime(remoteIdentifier, timeStamp); @@ -493,9 +514,9 @@ class ReplicatedChronicleMap extends VanillaChronicleMap public K key(@NotNull Bytes entry, K usingKey) { final long start = entry.position(); try { - // keyLen - entry.readStopBit(); - return keyMarshaller.read(entry, usingKey); + long keySize = keySizeMarshaller.readSize(entry); + ThreadLocalCopies copies = keyReaderProvider.getCopies(null); + return keyReaderProvider.get(copies, originalKeyReader).read(entry, keySize); } finally { entry.position(start); } @@ -506,9 +527,7 @@ class ReplicatedChronicleMap extends VanillaChronicleMap public V value(@NotNull Bytes entry, V usingValue) { final long start = entry.position(); try { - - // keyLen - entry.skip(entry.readStopBit()); + entry.skip(keySizeMarshaller.readSize(entry)); //timeStamp entry.readLong(); @@ -519,15 +538,17 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } final boolean isDeleted = entry.readBoolean(); - long valueLen; + long valueSize; if (!isDeleted) { - valueLen = entry.readStopBit(); - assert valueLen > 0; + valueSize = valueSizeMarshaller.readSize(entry); + assert valueSize > 0; } else { return null; } alignment.alignPositionAddr(entry); - return valueMarshaller.read(entry, usingValue); + ThreadLocalCopies copies = valueReaderProvider.getCopies(null); + BytesReader valueReader = valueReaderProvider.get(copies, originalValueReader); + return valueReader.read(entry, valueSize, usingValue); } finally { entry.position(start); } @@ -537,13 +558,13 @@ class ReplicatedChronicleMap extends VanillaChronicleMap public boolean wasRemoved(@NotNull Bytes entry) { final long start = entry.position(); try { - return entry.readBoolean(entry.readStopBit() + 10); + return entry.readBoolean(keySizeMarshaller.readSize(entry) + 10); } finally { entry.position(start); } } - class Segment extends VanillaChronicleMap.Segment { + class Segment extends VanillaChronicleMap.Segment { private volatile IntIntMultiMap hashLookupLiveAndDeleted; private volatile IntIntMultiMap hashLookupLiveOnly; @@ -563,10 +584,10 @@ class ReplicatedChronicleMap extends VanillaChronicleMap return hashLookupLiveOnly; } - private long entrySize(long keyLen, long valueLen) { + private long entrySize(long keySize, long valueSize) { long result = alignment.alignAddr(metaDataBytes + - expectedStopBits(keyLen) + keyLen + 10 + - expectedStopBits(valueLen)) + valueLen; + keySizeMarshaller.sizeEncodingSize(keySize) + keySize + 10 + + valueSizeMarshaller.sizeEncodingSize(valueSize)) + valueSize; // replication enforces that the entry size will never be larger than an unsigned short if (result > MAX_UNSIGNED_SHORT) throw new IllegalStateException("ENTRY WRITE_BUFFER_SIZE TOO LARGE : Replicated " + @@ -581,24 +602,28 @@ class ReplicatedChronicleMap extends VanillaChronicleMap /** * @see VanillaChronicleMap.Segment#acquire */ - V acquire(Bytes keyBytes, K key, V usingValue, int hash2, boolean create, long timestamp) { + V acquire(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, + K key, V usingValue, int hash2, boolean create, long timestamp) { lock(); try { + long keySize = metaKeyWriter.size(keyWriter, key); MultiStoreBytes entry = tmpBytes; - long offset = searchKey(keyBytes, hash2, entry, hashLookupLiveOnly); + long offset = searchKey(keyWriter, metaKeyWriter, key, keySize, hash2, entry, + hashLookupLiveOnly); if (offset >= 0) { // skip the timestamp, identifier and is deleted flag entry.skip(10); - return onKeyPresentOnAcquire(key, usingValue, offset, entry); + return onKeyPresentOnAcquire(copies, key, usingValue, offset, entry); } else { boolean usingValuePassed = usingValue != null; - usingValue = tryObtainUsingValueOnAcquire(keyBytes, key, usingValue, create); + usingValue = tryObtainUsingValueOnAcquire(key, usingValue, create); if (usingValue != null) { // see VanillaChronicleMap.Segment.acquire() for explanation // why `usingValue` is `create`. - offset = putEntryOnAcquire(keyBytes, hash2, usingValue, create, timestamp); + offset = putEntryOnAcquire(copies, metaKeyWriter, keyWriter, key, keySize, + hash2, usingValue, create, timestamp); incrementSize(); if (usingValuePassed || !create) notifyPut(offset, true, key, usingValue, posFromOffset(offset)); @@ -612,10 +637,12 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } } - private long putEntryOnAcquire(Bytes keyBytes, int hash2, V value, boolean usingValue, + private long putEntryOnAcquire(ThreadLocalCopies copies, + MKI metaKeyWriter, KI keyWriter, K key, + long keySize, int hash2, V value, boolean usingValue, long timestamp) { - return putEntry(keyBytes, hash2, value, usingValue, localIdentifier, - timestamp, hashLookupLiveOnly); + return putEntry(copies, metaKeyWriter, keyWriter, key, keySize, hash2, value, + usingValue, localIdentifier, timestamp, hashLookupLiveOnly); } @@ -626,16 +653,16 @@ class ReplicatedChronicleMap extends VanillaChronicleMap final long timestamp, final byte identifier) { lock(); try { - long keyLen = keyBytes.remaining(); + long keySize = keyBytes.remaining(); hashLookupLiveAndDeleted.startSearch(hash2); for (int pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); - if (!keyEquals(keyBytes, keyLen, entry)) + if (!remoteKeyEquals(keyBytes, keySize, entry)) continue; // key is found - entry.skip(keyLen); + entry.skip(keySize); long timeStampPos = entry.position(); @@ -677,17 +704,17 @@ class ReplicatedChronicleMap extends VanillaChronicleMap lock(); try { // inBytes position and limit correspond to the key - final long keyLen = inBytes.remaining(); + final long keySize = inBytes.remaining(); hashLookupLiveAndDeleted.startSearch(hash2); for (int pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); - if (!keyEquals(inBytes, keyLen, entry)) + if (!remoteKeyEquals(inBytes, keySize, entry)) continue; // key is found - entry.skip(keyLen); + entry.skip(keySize); final long timeStampPos = entry.positionAddr(); @@ -704,15 +731,15 @@ class ReplicatedChronicleMap extends VanillaChronicleMap // deleted flag entry.writeBoolean(false); - long valueLenPos = entry.position(); - long valueLen = readValueLen(entry); - long entryEndAddr = entry.positionAddr() + valueLen; + long valueSizePos = entry.position(); + long valueSize = readValueSize(entry); + long entryEndAddr = entry.positionAddr() + valueSize; // write the value inBytes.limit(valueLimit); inBytes.position(valuePos); - putValue(pos, offset, entry, valueLenPos, entryEndAddr, inBytes, null, true, + putValue(pos, offset, entry, valueSizePos, entryEndAddr, null, null, inBytes, hashLookupLiveAndDeleted); if (wasDeleted) { @@ -724,13 +751,13 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } // key is not found - long valueLen = valueLimit - valuePos; - int pos = alloc(inBlocks(entrySize(keyLen, valueLen))); + long valueSize = valueLimit - valuePos; + int pos = alloc(inBlocks(entrySize(keySize, valueSize))); long offset = offsetFromPos(pos); clearMetaData(offset); NativeBytes entry = entry(offset); - entry.writeStopBit(keyLen); + keySizeMarshaller.writeSize(entry, keySize); // write the key entry.write(inBytes); @@ -739,7 +766,7 @@ class ReplicatedChronicleMap extends VanillaChronicleMap entry.writeByte(identifier); entry.writeBoolean(false); - entry.writeStopBit(valueLen); + valueSizeMarshaller.writeSize(entry, valueSize); alignment.alignPositionAddr(entry); // write the value @@ -758,21 +785,26 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } } - V put(Bytes keyBytes, K key, V value, int hash2, boolean replaceIfPresent, + private boolean remoteKeyEquals(Bytes keyBytes, long keySize, Bytes entry) { + return keySize == keySizeMarshaller.readSize(entry) && entry.startsWith(keyBytes); + } + + V put(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, + K key, V value, int hash2, boolean replaceIfPresent, final byte identifier, final long timestamp) { lock(); try { IntIntMultiMap hashLookup = hashLookupLiveAndDeleted; - long keyLen = keyBytes.remaining(); + long keySize = metaKeyWriter.size(keyWriter, key); hashLookup.startSearch(hash2); int pos; while ((pos = hashLookup.nextPos()) >= 0) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); - if (!keyEquals(keyBytes, keyLen, entry)) + if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) continue; // key is found - entry.skip(keyLen); + entry.skip(keySize); final long timeStampPos = entry.positionAddr(); @@ -791,8 +823,8 @@ class ReplicatedChronicleMap extends VanillaChronicleMap // deleted flag entry.writeBoolean(false); - final V prevValue = replaceValueOnPut(key, value, entry, pos, offset, - !wasDeleted && !putReturnsNull, hashLookup); + final V prevValue = replaceValueOnPut(copies, key, value, entry, pos, + offset, !wasDeleted && !putReturnsNull, hashLookup); if (wasDeleted) { // remove() would have got rid of this so we have to add it back in @@ -804,12 +836,13 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } } else { // we wont replaceIfPresent and the entry is not deleted - return putReturnsNull ? null : readValue(entry, null); + return putReturnsNull ? null : readValue(copies, entry, null); } } // key is not found - long offset = putEntry(keyBytes, hash2, value, false, identifier, timestamp, hashLookup); + long offset = putEntry(copies, metaKeyWriter, keyWriter, key, keySize, hash2, value, + false, identifier, timestamp, hashLookup); incrementSize(); notifyPut(offset, true, key, value, posFromOffset(offset)); return null; @@ -848,55 +881,61 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } /** - * Puts entry. If {@code value} implements {@link net.openhft.lang.model.Byteable} interface and + * Puts entry. If {@code value} implements {@link Byteable} interface and * {@code usingValue} is {@code true}, the value is backed with the bytes of this entry. * - * @param keyBytes serialized key * @param hash2 a hash of the {@code keyBytes}. Caller was searching for the key in the * {@code searchedHashLookup} using this hash. * @param value the value to put * @param usingValue {@code true} if the value should be backed with the bytes of the entry, - * if it implements {@link net.openhft.lang.model.Byteable} interface, + * if it implements {@link Byteable} interface, * {@code false} if it should put itself * @param identifier the identifier of the outer CHM node * @param timestamp the timestamp when the entry was put (this could be later if it was a * remote put) this method is called only from usual put or acquire * @param searchedHashLookup the hash lookup that used to find the entry based on the key * @return offset of the written entry in the Segment bytes - * @see VanillaChronicleMap.Segment#putEntry(Bytes, Object, boolean) + * @see VanillaChronicleMap.Segment#putEntry */ - private long putEntry(Bytes keyBytes, int hash2, V value, boolean usingValue, - final int identifier, final long timestamp, + private long putEntry(ThreadLocalCopies copies, MKI metaKeyWriter, + KI keyWriter, K key, long keySize, int hash2, V value, + boolean usingValue, final int identifier, final long timestamp, IntIntMultiMap searchedHashLookup) { - long keyLen = keyBytes.remaining(); // "if-else polymorphism" is not very beautiful, but allows to // reuse the rest code of this method and doesn't hurt performance. boolean byteableValue = usingValue && value instanceof Byteable; - long valueLen; - Bytes valueBytes = null; + long valueSize; + MetaBytesWriter metaValueWriter = null; + VW valueWriter = null; Byteable valueAsByteable = null; if (!byteableValue) { - valueBytes = getValueAsBytes(value); - valueLen = valueBytes.remaining(); + copies = valueWriterProvider.getCopies(copies); + valueWriter = valueWriterProvider.get(copies, originalValueWriter); + copies = metaValueWriterProvider.getCopies(copies); + metaValueWriter = metaValueWriterProvider.get( + copies, originalMetaValueWriter, valueWriter, value); + valueSize = metaValueWriter.size(valueWriter, value); } else { valueAsByteable = (Byteable) value; - valueLen = valueAsByteable.maxSize(); + valueSize = valueAsByteable.maxSize(); } - long entrySize = entrySize(keyLen, valueLen); + long entrySize = entrySize(keySize, valueSize); int pos = alloc(inBlocks(entrySize)); long offset = offsetFromPos(pos); clearMetaData(offset); NativeBytes entry = entry(offset); - entry.writeStopBit(keyLen); - entry.write(keyBytes); + keySizeMarshaller.writeSize(entry, keySize); + metaKeyWriter.write(keyWriter, entry, key); + entry.writeLong(timestamp); entry.writeByte(identifier); entry.writeBoolean(false); - writeValueOnPutEntry(valueLen, valueBytes, valueAsByteable, entry); + writeValueOnPutEntry(valueSize, metaValueWriter, valueWriter, value, valueAsByteable, + entry); // we have to add it both to the live if (searchedHashLookup == hashLookupLiveAndDeleted) { @@ -911,25 +950,24 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } /** - * @see VanillaChronicleMap.Segment#remove(net.openhft.lang.io.Bytes, Object, Object, int) + * @see VanillaChronicleMap.Segment#remove */ - public V remove(Bytes keyBytes, K key, V expectedValue, int hash2, - final long timestamp, final byte identifier) { + V remove(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, + K key, V expectedValue, int hash2, + final long timestamp, final byte identifier) { assert identifier > 0; lock(); try { - long keyLen = keyBytes.remaining(); - + long keySize = metaKeyWriter.size(keyWriter, key); final IntIntMultiMap multiMap = hashLookupLiveAndDeleted; - multiMap.startSearch(hash2); for (int pos; (pos = multiMap.nextPos()) >= 0; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); - if (!keyEquals(keyBytes, keyLen, entry)) + if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) continue; // key is found - entry.skip(keyLen); + entry.skip(keySize); long timeStampPos = entry.position(); if (shouldIgnore(entry, timestamp, identifier)) { @@ -951,9 +989,9 @@ class ReplicatedChronicleMap extends VanillaChronicleMap return null; } - long valueLen = readValueLen(entry); + long valueSize = readValueSize(entry); V valueRemoved = expectedValue != null || !removeReturnsNull - ? readValue(entry, null, valueLen) : null; + ? readValue(copies, entry, null, valueSize) : null; if (expectedValue != null && !expectedValue.equals(valueRemoved)) return null; @@ -982,29 +1020,29 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } /** - * @see VanillaChronicleMap.Segment#remove(net.openhft.lang.io.Bytes, Object, Object, int) + * @see VanillaChronicleMap.Segment#replace */ - public V replace(Bytes keyBytes, K key, V expectedValue, V newValue, int hash2, - long timestamp) { + V replace(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, + K key, V expectedValue, V newValue, int hash2, long timestamp) { lock(); try { - long keyLen = keyBytes.remaining(); + long keySize = metaKeyWriter.size(keyWriter, key); hashLookupLiveOnly.startSearch(hash2); for (int pos; (pos = hashLookupLiveOnly.nextPos()) >= 0; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); - if (!keyEquals(keyBytes, keyLen, entry)) + if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) continue; // key is found - entry.skip(keyLen); + entry.skip(keySize); if (shouldIgnore(entry, timestamp, localIdentifier)) return null; // skip the is deleted flag entry.skip(1); - return onKeyPresentOnReplace(key, expectedValue, newValue, pos, offset, entry, - hashLookupLiveOnly); + return onKeyPresentOnReplace(copies, key, expectedValue, newValue, pos, offset, + entry, hashLookupLiveOnly); } // key is not found return null; @@ -1033,8 +1071,8 @@ class ReplicatedChronicleMap extends VanillaChronicleMap @Override public void accept(int hash, int pos) { final NativeBytes entry = entry(offsetFromPos(pos)); - long keyLen = entry.readStopBit(); - entry.skip(keyLen); + long keySize = keySizeMarshaller.readSize(entry); + entry.skip(keySize); final long entryTimestamp = entry.readLong(); @@ -1052,13 +1090,19 @@ class ReplicatedChronicleMap extends VanillaChronicleMap @Override public Entry getEntry(long pos) { bytes.position(offsetFromPos(pos) + metaDataBytes); - bytes.readStopBit(); - K key = keyMarshaller.read(bytes); + + long keySize = keySizeMarshaller.readSize(bytes); + ThreadLocalCopies copies = keyReaderProvider.getCopies(null); + K key = keyReaderProvider.get(copies, originalKeyReader).read(bytes, keySize); + long timestamp = bytes.readLong(); bytes.skip(2L); // identifier and isDeleted flag - bytes.readStopBit(); + + long valueSize = valueSizeMarshaller.readSize(bytes); alignment.alignPositionAddr(bytes); - V value = valueMarshaller.read(bytes); + copies = valueReaderProvider.getCopies(copies); + V value = valueReaderProvider.get(copies, originalValueReader).read(bytes, valueSize); + return new TimestampTrackingEntry(key, value, timestamp); } @@ -1090,7 +1134,7 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } } - class EntryIterator extends VanillaChronicleMap.EntryIterator { + class EntryIterator extends VanillaChronicleMap.EntryIterator { @Override void removePresent(VanillaChronicleMap.Segment seg, int pos) { @SuppressWarnings("unchecked") @@ -1100,17 +1144,23 @@ class ReplicatedChronicleMap extends VanillaChronicleMap final NativeBytes entry = segment.entry(offset); final long limit = entry.limit(); - final long keyLen = entry.readStopBit(); + final long keySize = keySizeMarshaller.readSize(entry); long keyPosition = entry.position(); - entry.skip(keyLen); + entry.skip(keySize); long timestamp = entry.readLong(); entry.position(keyPosition); if (timestamp > ((TimestampTrackingEntry) returnedEntry).timestamp) { // The entry was updated after being returned from iterator.next() // Check that it is still the entry with the same key K key = returnedEntry.getKey(); - DirectBytes returnedKeyBytes = getKeyAsBytes(key); - if (returnedKeyBytes.remaining() != keyLen || !entry.startsWith(returnedKeyBytes)) { + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyWriter = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyWriter = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); + long returnedKeySize = metaKeyWriter.size(keyWriter, key); + if (returnedKeySize != keySize || + !metaKeyWriter.startsWith(keyWriter, entry, key)) { // The case: // 1. iterator.next() - thread 1 // 2. map.put() which cause relocation of the key, returned above - thread 2 @@ -1121,14 +1171,12 @@ class ReplicatedChronicleMap extends VanillaChronicleMap return; } } - entry.limit(entry.position() + keyLen); - final int segmentHash = hasher.segmentHash(hash(entry)); - entry.limit(limit); + final int segmentHash = segmentHash(hash(entry, keyPosition, keyPosition + keySize)); segment.hashLookupLiveOnly.remove(segmentHash, pos); segment.decrementSize(); - entry.skip(keyLen); + entry.skip(keySize); entry.writeLong(timeProvider.currentTimeMillis()); entry.writeByte(localIdentifier); entry.writeBoolean(true); @@ -1137,7 +1185,7 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } } - class EntrySet extends VanillaChronicleMap.EntrySet { + class EntrySet extends VanillaChronicleMap.EntrySet { @NotNull @Override public Iterator> iterator() { @@ -1252,9 +1300,9 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } @Override - public V onGetMissing(ChronicleMap map, Bytes keyBytes, K key, + public V onGetMissing(ChronicleMap map, K key, V usingValue) { - return nextListener.onGetMissing(map, keyBytes, key, usingValue); + return nextListener.onGetMissing(map, key, usingValue); } @Override diff --git a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java new file mode 100644 index 000000000..ca6743b2f --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java @@ -0,0 +1,400 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.map.serialization.*; +import net.openhft.chronicle.map.serialization.impl.*; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.*; +import net.openhft.lang.io.serialization.impl.*; +import net.openhft.lang.model.Byteable; +import net.openhft.lang.model.DataValueClasses; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +import java.io.Externalizable; +import java.io.Serializable; + +import static net.openhft.chronicle.map.Objects.hash; +import static net.openhft.chronicle.map.serialization.SizeMarshallers.stopBit; + +final class SerializationBuilder implements Cloneable { + + private static boolean marshallerUseFactory(Class c) { + return Byteable.class.isAssignableFrom(c) || + BytesMarshallable.class.isAssignableFrom(c) || + Externalizable.class.isAssignableFrom(c); + } + + enum Role {KEY, VALUE} + private enum CopyingInterop {FROM_MARSHALLER, FROM_WRITER} + + private final Serializable bufferIdentity; + final Class eClass; + private SizeMarshaller sizeMarshaller = stopBit(); + private BytesReader reader; + private Object interop; + private CopyingInterop copyingInterop = null; + private MetaBytesInterop metaInterop; + private MetaProvider metaInteropProvider; + private ObjectFactory factory; + + @SuppressWarnings("unchecked") + SerializationBuilder(Class eClass, Role role) { + this.bufferIdentity = role; + this.eClass = eClass; + Class classForMarshaller = marshallerUseFactory(eClass) && eClass.isInterface() ? + DataValueClasses.directClassFor(eClass) : eClass; + + ObjectFactory factory = null; + if (role == Role.VALUE) { + factory = marshallerUseFactory(eClass) ? + new AllocateInstanceObjectFactory( + eClass.isInterface() ? classForMarshaller : eClass) : + NullObjectFactory.INSTANCE; + } + + if (Byteable.class.isAssignableFrom(eClass)) { + agileMarshaller(ByteableMarshaller.of((Class) classForMarshaller), factory); + } else if (eClass == CharSequence.class || eClass == String.class) { + reader((BytesReader) CharSequenceReader.of(), factory); + writer((BytesWriter) CharSequenceWriter.instance()); + } else if (eClass == Long.class) { + agileMarshaller((AgileBytesMarshaller) LongMarshaller.INSTANCE, factory); + } else if (eClass == byte[].class) { + reader((BytesReader) ByteArrayMarshaller.INSTANCE, factory); + interop((BytesInterop) ByteArrayMarshaller.INSTANCE); + } else { + marshaller(chooseMarshaller(eClass, classForMarshaller), factory); + } + } + + @SuppressWarnings("unchecked") + private BytesMarshaller chooseMarshaller(Class eClass, Class classForMarshaller) { + if (BytesMarshallable.class.isAssignableFrom(eClass)) + return new BytesMarshallableMarshaller(classForMarshaller); + if (Externalizable.class.isAssignableFrom(eClass)) + return new ExternalizableMarshaller(classForMarshaller); + if (eClass == Integer.class) + return (BytesMarshaller) IntegerMarshaller.INSTANCE; + if (eClass == Double.class) + return (BytesMarshaller) DoubleMarshaller.INSTANCE; + return SerializableMarshaller.INSTANCE; + } + + private SerializationBuilder copyingInterop(CopyingInterop copyingInterop) { + this.copyingInterop = copyingInterop; + return this; + } + + public SerializationBuilder agileMarshaller(AgileBytesMarshaller agileBytesMarshaller, + ObjectFactory factory) { + if (factory == null) factory = this.factory; + return sizeMarshaller(agileBytesMarshaller) + .reader(agileBytesMarshaller, factory) + .interop(agileBytesMarshaller); + } + + public SerializationBuilder interop(BytesInterop interop) { + return copyingInterop(null) + .setInterop(interop) + .metaInterop(DelegatingMetaBytesInterop.>instance()) + .metaInteropProvider(DelegatingMetaBytesInteropProvider + .>instance()); + } + + public SerializationBuilder writer(BytesWriter writer) { + return copyingInterop(CopyingInterop.FROM_WRITER) + .sizeMarshaller(stopBit()) + .setInterop(writer) + .metaInterop(CopyingMetaBytesInterop + .>forBytesWriter(bufferIdentity)); + } + + public SerializationBuilder marshaller(BytesMarshaller marshaller, + ObjectFactory factory) { + if (factory == null) factory = this.factory; + return copyingInterop(CopyingInterop.FROM_MARSHALLER) + .sizeMarshaller(stopBit()) + .reader(BytesReaders.fromBytesMarshaller(marshaller), factory) + .setInterop(marshaller) + .metaInterop(CopyingMetaBytesInterop + .>forBytesMarshaller(bufferIdentity)); + } + + public SerializationBuilder maxSize(long maxSize) { + boolean mutable = eClass != String.class; // for example + if (copyingInterop == CopyingInterop.FROM_MARSHALLER) { + metaInteropProvider(CopyingMetaBytesInterop + .>providerForBytesMarshaller(mutable, maxSize)); + } else if (copyingInterop == CopyingInterop.FROM_WRITER) { + metaInteropProvider(CopyingMetaBytesInterop + .>providerForBytesWriter(mutable, maxSize)); + } + return this; + } + + public SizeMarshaller sizeMarshaller() { + return sizeMarshaller; + } + + public SerializationBuilder sizeMarshaller(SizeMarshaller sizeMarshaller) { + this.sizeMarshaller = sizeMarshaller; + return this; + } + + public BytesReader reader() { + return reader; + } + + public SerializationBuilder reader(BytesReader reader, ObjectFactory factory) { + this.reader = reader; + if (factory != null) + this.factory = factory; + return this; + + } + + public Object interop() { + return interop; + } + + private SerializationBuilder setInterop(Object interop) { + this.interop = interop; + return this; + } + + public MetaBytesInterop metaInterop() { + return metaInterop; + } + + private SerializationBuilder metaInterop(MetaBytesInterop metaInterop) { + this.metaInterop = metaInterop; + return this; + } + + @SuppressWarnings("unchecked") + public MetaProvider> metaInteropProvider() { + return (MetaProvider>) metaInteropProvider; + } + + private SerializationBuilder metaInteropProvider( + MetaProvider metaInteropProvider) { + this.metaInteropProvider = metaInteropProvider; + return this; + } + + public ObjectFactory factory() { + return factory; + } + + @SuppressWarnings("unchecked") + public SerializationBuilder factory(ObjectFactory factory) { + if (!marshallerUseFactory(eClass)) { + throw new IllegalStateException("Default marshaller for " + eClass + + " value don't use object factory"); + } else if (interop instanceof ByteableMarshaller) { + if (factory instanceof AllocateInstanceObjectFactory) { + agileMarshaller(ByteableMarshaller.of((Class) eClass), factory); + } else { + agileMarshaller(ByteableMarshaller.of((Class) eClass, (ObjectFactory) factory), + factory); + } + } else if (interop instanceof BytesMarshallableMarshaller) { + if (factory instanceof AllocateInstanceObjectFactory) { + interop = new BytesMarshallableMarshaller( + ((AllocateInstanceObjectFactory) factory).allocatedClass()); + } else { + interop = new BytesMarshallableMarshallerWithCustomFactory( + ((BytesMarshallableMarshaller) interop).marshaledClass(), + factory + ); + } + } else if (interop instanceof ExternalizableMarshaller) { + if (factory instanceof AllocateInstanceObjectFactory) { + interop = new ExternalizableMarshaller( + ((AllocateInstanceObjectFactory) factory).allocatedClass()); + } else { + interop = new ExternalizableMarshallerWithCustomFactory( + ((ExternalizableMarshaller) interop).marshaledClass(), + factory + ); + } + } else { + // interop is custom, it is user's responsibility to use the same factory inside + // marshaller and standalone + throw new IllegalStateException( + "Change the value factory simultaneously with marshaller " + + "using *AndFactory() method"); + } + this.factory = factory; + return this; + } + + @Override + public SerializationBuilder clone() { + try { + return (SerializationBuilder) super.clone(); + } catch (CloneNotSupportedException e) { + throw new AssertionError(e); + } + } + + private enum IntegerMarshaller implements BytesMarshaller { + INSTANCE; + + @Override + public void write(Bytes bytes, Integer v) { + bytes.writeInt(v); + } + + @Nullable + @Override + public Integer read(Bytes bytes) { + return bytes.readInt(); + } + + @Nullable + @Override + public Integer read(Bytes bytes, @Nullable Integer v) { + return bytes.readInt(); + } + } + + private enum DoubleMarshaller implements BytesMarshaller { + INSTANCE; + + @Override + public void write(Bytes bytes, Double v) { + bytes.writeDouble(v); + } + + @Nullable + @Override + public Double read(Bytes bytes) { + return bytes.readDouble(); + } + + @Nullable + @Override + public Double read(Bytes bytes, @Nullable Double v) { + return bytes.readDouble(); + } + } + + + private enum SerializableMarshaller implements BytesMarshaller { + INSTANCE; + + @Override + public void write(Bytes bytes, Object obj) { + bytes.writeObject(obj); + } + + @Nullable + @Override + public Object read(Bytes bytes) { + return bytes.readObject(); + } + + @Nullable + @Override + public Object read(Bytes bytes, @Nullable Object obj) { + return bytes.readInstance(null, obj); + } + } + + private static class BytesMarshallableMarshallerWithCustomFactory + extends BytesMarshallableMarshaller { + private static final long serialVersionUID = 0L; + + @NotNull + private final ObjectFactory factory; + + BytesMarshallableMarshallerWithCustomFactory(@NotNull Class tClass, + @NotNull ObjectFactory factory) { + super(tClass); + this.factory = factory; + } + + @NotNull + @Override + protected T getInstance() throws Exception { + return factory.create(); + } + + @Override + public boolean equals(Object obj) { + if (obj == null || obj.getClass() != getClass()) + return false; + BytesMarshallableMarshallerWithCustomFactory that = + (BytesMarshallableMarshallerWithCustomFactory) obj; + return that.marshaledClass() == marshaledClass() && that.factory.equals(this.factory); + } + + @Override + public int hashCode() { + return hash(marshaledClass(), factory); + } + + @Override + public String toString() { + return getClass().getSimpleName() + "{marshaledClass=" + marshaledClass() + + ",factory=" + factory + "}"; + } + } + + private static class ExternalizableMarshallerWithCustomFactory + extends ExternalizableMarshaller { + private static final long serialVersionUID = 0L; + + @NotNull + private final ObjectFactory factory; + + ExternalizableMarshallerWithCustomFactory(@NotNull Class tClass, + @NotNull ObjectFactory factory) { + super(tClass); + this.factory = factory; + } + + @NotNull + @Override + protected T getInstance() throws Exception { + return factory.create(); + } + + @Override + public boolean equals(Object obj) { + if (obj == null || obj.getClass() != getClass()) + return false; + ExternalizableMarshallerWithCustomFactory that = + (ExternalizableMarshallerWithCustomFactory) obj; + return that.marshaledClass() == marshaledClass() && that.factory.equals(this.factory); + } + + @Override + public int hashCode() { + return hash(marshaledClass(), factory); + } + + @Override + public String toString() { + return getClass().getSimpleName() + "{marshaledClass=" + marshaledClass() + + ",factory=" + factory + "}"; + } + } +} diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index cf990baff..5aaae93ca 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -16,18 +16,17 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.map.serialization.*; import net.openhft.lang.Maths; import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.collection.SingleThreadedDirectBitSet; import net.openhft.lang.io.*; -import net.openhft.lang.io.serialization.BytesMarshallable; -import net.openhft.lang.io.serialization.BytesMarshaller; -import net.openhft.lang.io.serialization.ObjectFactory; -import net.openhft.lang.io.serialization.ObjectSerializer; +import net.openhft.lang.io.serialization.*; import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory; import net.openhft.lang.model.Byteable; +import net.openhft.chronicle.map.threadlocal.Provider; +import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; -import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,25 +40,40 @@ import java.util.*; import java.util.concurrent.TimeUnit; import static java.lang.Thread.currentThread; -import static net.openhft.chronicle.map.VanillaChronicleMap.Hasher.hash; -class VanillaChronicleMap extends AbstractMap +class VanillaChronicleMap, + V, VW, MVW extends MetaBytesWriter> extends AbstractMap implements ChronicleMap, Serializable { - private static final long serialVersionUID = 0L; + private static final long serialVersionUID = 1L; private static final Logger LOG = LoggerFactory.getLogger(VanillaChronicleMap.class); /** * Because DirectBitSet implementations couldn't find more than 64 continuous clear or set bits. */ - private static final int MAX_ENTRY_OVERSIZE_FACTOR = 64; + static final int MAX_ENTRY_OVERSIZE_FACTOR = 64; private static final int SEGMENT_HEADER = 64; final Class kClass; - final BytesMarshaller keyMarshaller; - final BytesMarshaller valueMarshaller; + final SizeMarshaller keySizeMarshaller; + final BytesReader originalKeyReader; + transient Provider> keyReaderProvider; + final KI originalKeyInterop; + transient Provider keyInteropProvider; + final MKI originalMetaKeyInterop; + final MetaProvider metaKeyInteropProvider; + + final Class vClass; + final SizeMarshaller valueSizeMarshaller; + final BytesReader originalValueReader; + transient Provider> valueReaderProvider; + final VW originalValueWriter; + transient Provider valueWriterProvider; + final MVW originalMetaValueWriter; + final MetaProvider metaValueWriterProvider; + final ObjectFactory valueFactory; + final int metaDataBytes; - final Hasher hasher; // private final int replicas; final int entrySize; final Alignment alignment; @@ -70,10 +84,8 @@ class VanillaChronicleMap extends AbstractMap // rather than as returning the Object can be expensive for something you probably don't use. final boolean putReturnsNull; final boolean removeReturnsNull; - private final int bufferAllocationFactor; private final ObjectSerializer objectSerializer; - private final Class vClass; - private final ObjectFactory valueFactory; + private final long lockTimeOutNS; private final MapErrorListener errorListener; transient Segment[] segments; // non-final for close() @@ -81,17 +93,31 @@ class VanillaChronicleMap extends AbstractMap transient BytesStore ms; transient long headerSize; transient Set> entrySet; - private transient ThreadLocal localBufferForKeys; - private transient ThreadLocal localBufferForValues; - public VanillaChronicleMap(ChronicleMapBuilder builder) - throws IOException { - bufferAllocationFactor = figureBufferAllocationFactor(builder); - this.kClass = builder.keyClass; - this.keyMarshaller = builder.keyMarshaller(); - this.vClass = builder.valueClass; - this.valueMarshaller = builder.valueMarshaller(); - this.valueFactory = builder.valueFactory(); + private int bits; + private int mask; + + public VanillaChronicleMap(ChronicleMapBuilder builder) throws IOException { + + SerializationBuilder keyBuilder = builder.keyBuilder; + kClass = keyBuilder.eClass; + keySizeMarshaller = keyBuilder.sizeMarshaller(); + originalKeyReader = keyBuilder.reader(); + + originalKeyInterop = (KI) keyBuilder.interop(); + + originalMetaKeyInterop = (MKI) keyBuilder.metaInterop(); + metaKeyInteropProvider = (MetaProvider) keyBuilder.metaInteropProvider(); + + SerializationBuilder valueBuilder = builder.valueBuilder; + vClass = valueBuilder.eClass; + valueSizeMarshaller = valueBuilder.sizeMarshaller(); + originalValueReader = valueBuilder.reader(); + + originalValueWriter = (VW) valueBuilder.interop(); + originalMetaValueWriter = (MVW) valueBuilder.metaInterop(); + metaValueWriterProvider = (MetaProvider) valueBuilder.metaInteropProvider(); + valueFactory = valueBuilder.factory(); lockTimeOutNS = builder.lockTimeOut(TimeUnit.NANOSECONDS); @@ -109,30 +135,18 @@ class VanillaChronicleMap extends AbstractMap this.metaDataBytes = builder.metaDataBytes(); this.eventListener = builder.eventListener(); - int hashMask = useSmallMultiMaps() ? 0xFFFF : ~0; - this.hasher = new Hasher(actualSegments, hashMask); + this.mask = useSmallMultiMaps() ? 0xFFFF : ~0; + this.bits = Maths.intLog2(actualSegments); initTransients(); } - private static int figureBufferAllocationFactor(ChronicleMapBuilder builder) { - // if expected map size is about 1000, seems rather wasteful to allocate - // key and value serialization buffers each x64 of expected entry size.. - return (int) Math.min(Math.max(2L, builder.entries() >> 10), - MAX_ENTRY_OVERSIZE_FACTOR); + int segmentHash(long hash) { + return (int) (hash >>> bits) & mask; } - /** - * @param size positive number - * @return number of bytes taken by {@link net.openhft.lang.io.AbstractBytes#writeStopBit(long)} applied - * to {@code size} - */ - static int expectedStopBits(long size) { - if (size <= 127) - return 1; - // numberOfLeadingZeros is cheap intrinsic on modern CPUs - // integral division is not... but there is no choice - return ((70 - Long.numberOfLeadingZeros(size)) / 7); + int getSegment(long hash) { + return (int) (hash & (actualSegments - 1)); } /** @@ -143,11 +157,15 @@ class VanillaChronicleMap extends AbstractMap } void initTransients() { + keyReaderProvider = Provider.of((Class) originalKeyReader.getClass()); + keyInteropProvider = Provider.of((Class) originalKeyInterop.getClass()); + + valueReaderProvider = Provider.of((Class) originalValueReader.getClass()); + valueWriterProvider = Provider.of((Class) originalValueWriter.getClass()); + @SuppressWarnings("unchecked") Segment[] ss = (Segment[]) Array.newInstance(segmentType(), actualSegments); this.segments = ss; - localBufferForKeys = new ThreadLocal(); - localBufferForValues = new ThreadLocal(); } Class segmentType() { @@ -275,30 +293,6 @@ class VanillaChronicleMap extends AbstractMap return eventListener; } - private DirectBytes acquireBufferForKey() { - DirectBytes buffer = localBufferForKeys.get(); - if (buffer == null) { - buffer = new DirectStore(ms.objectSerializer(), - entrySize * bufferAllocationFactor, false).bytes(); - localBufferForKeys.set(buffer); - } else { - buffer.clear(); - } - return buffer; - } - - private DirectBytes acquireBufferForValue() { - DirectBytes buffer = localBufferForValues.get(); - if (buffer == null) { - buffer = new DirectStore(ms.objectSerializer(), - entrySize * bufferAllocationFactor, false).bytes(); - localBufferForValues.set(buffer); - } else { - buffer.clear(); - } - return buffer; - } - void checkKey(Object key) { if (!kClass.isInstance(key)) { // key.getClass will cause NPE exactly as needed @@ -327,26 +321,16 @@ class VanillaChronicleMap extends AbstractMap private V put0(K key, V value, boolean replaceIfPresent) { checkKey(key); checkValue(value); - Bytes keyBytes = getKeyAsBytes(key); - long hash = hash(keyBytes); - int segmentNum = hasher.getSegment(hash); - int segmentHash = hasher.segmentHash(hash); - return segments[segmentNum].put(keyBytes, key, value, segmentHash, replaceIfPresent); - } - - DirectBytes getKeyAsBytes(K key) { - DirectBytes buffer = acquireBufferForKey(); - keyMarshaller.write(buffer, key); - buffer.flip(); - return buffer; - } - - DirectBytes getValueAsBytes(V value) { - DirectBytes buffer = acquireBufferForValue(); - buffer.clear(); - valueMarshaller.write(buffer, value); - buffer.flip(); - return buffer; + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyInterop = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); + long hash = metaKeyInterop.hash(keyInterop, key); + int segmentNum = getSegment(hash); + int segmentHash = segmentHash(hash); + return segments[segmentNum].put(copies, metaKeyInterop, keyInterop, key, value, segmentHash, + replaceIfPresent); } @Override @@ -366,22 +350,31 @@ class VanillaChronicleMap extends AbstractMap V lookupUsing(K key, V value, boolean create) { checkKey(key); - Bytes keyBytes = getKeyAsBytes(key); - long hash = hash(keyBytes); - int segmentNum = hasher.getSegment(hash); - int segmentHash = hasher.segmentHash(hash); - return segments[segmentNum].acquire(keyBytes, key, value, segmentHash, create); + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyInterop = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); + long hash = metaKeyInterop.hash(keyInterop, key); + int segmentNum = getSegment(hash); + int segmentHash = segmentHash(hash); + return segments[segmentNum].acquire(copies, metaKeyInterop, keyInterop, key, value, + segmentHash, create); } @Override - public boolean containsKey(final Object key) { - checkKey(key); - Bytes keyBytes = getKeyAsBytes((K) key); - long hash = hash(keyBytes); - int segmentNum = hasher.getSegment(hash); - int segmentHash = hasher.segmentHash(hash); - - return segments[segmentNum].containsKey(keyBytes, segmentHash); + public boolean containsKey(final Object k) { + checkKey(k); + K key = (K) k; + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyInterop = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); + long hash = metaKeyInterop.hash(keyInterop, key); + int segmentNum = getSegment(hash); + int segmentHash = segmentHash(hash); + return segments[segmentNum].containsKey(keyInterop, metaKeyInterop, key, segmentHash); } @Override @@ -421,17 +414,23 @@ class VanillaChronicleMap extends AbstractMap * that of a maps.entry. If the {@param expectedValue} equals null then ( if there exists ) an entry whose * key equals {@param key} this is removed. * - * @param key the key of the entry to remove + * @param k the key of the entry to remove * @param expectedValue null if not required * @return true if and entry was removed */ - V removeIfValueIs(final Object key, final V expectedValue) { - checkKey(key); - Bytes keyBytes = getKeyAsBytes((K) key); - long hash = hash(keyBytes); - int segmentNum = hasher.getSegment(hash); - int segmentHash = hasher.segmentHash(hash); - return segments[segmentNum].remove(keyBytes, (K) key, expectedValue, segmentHash); + V removeIfValueIs(final Object k, final V expectedValue) { + checkKey(k); + K key = (K) k; + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyInterop = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); + long hash = metaKeyInterop.hash(keyInterop, key); + int segmentNum = getSegment(hash); + int segmentHash = segmentHash(hash); + return segments[segmentNum].remove(copies, metaKeyInterop, keyInterop, key, expectedValue, + segmentHash); } /** @@ -482,11 +481,16 @@ class VanillaChronicleMap extends AbstractMap final V newValue) { checkKey(key); checkValue(newValue); - Bytes keyBytes = getKeyAsBytes(key); - long hash = hash(keyBytes); - int segmentNum = hasher.getSegment(hash); - int segmentHash = hasher.segmentHash(hash); - return segments[segmentNum].replace(keyBytes, key, existingValue, newValue, segmentHash); + ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop); + copies = metaKeyInteropProvider.getCopies(copies); + MKI metaKeyInterop = + metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); + long hash = metaKeyInterop.hash(keyInterop, key); + int segmentNum = getSegment(hash); + int segmentHash = segmentHash(hash); + return segments[segmentNum].replace(copies, metaKeyInterop, keyInterop, key, existingValue, + newValue, segmentHash); } /** @@ -498,51 +502,20 @@ class VanillaChronicleMap extends AbstractMap } } - - static final class Hasher implements Serializable { - private static final long serialVersionUID = 0L; - private final int segments; - private final int bits; - private final int mask; - - Hasher(int segments, int mask) { - this.segments = segments; - this.bits = Maths.intLog2(segments); - this.mask = mask; - } - - static long hash(Bytes bytes) { - long h = 0; - long i = bytes.position(); - long limit = bytes.limit(); // clustering. - for (; i < limit - 7; i += 8) - h = 1011001110001111L * h + bytes.readLong(i); - for (; i < limit - 1; i += 2) - h = 101111 * h + bytes.readShort(i); - if (i < limit) - h = 2111 * h + bytes.readByte(i); - h *= 11018881818881011L; - h ^= (h >>> 41) ^ (h >>> 21); - //System.out.println(bytes + " => " + Long.toHexString(h)); - return h; - } - - int segmentHash(long hash) { - return (int) (hash >>> bits) & mask; - } - - int getSegment(long hash) { - return (int) (hash & (segments - 1)); - } + long readValueSize(Bytes entry) { + long valueSize = valueSizeMarshaller.readSize(entry); + alignment.alignPositionAddr(entry); + return valueSize; } // these methods should be package local, not public or private. class Segment implements SharedSegment { /* The entry format is - - stop-bit encoded length for key + - encoded length for key - bytes for the key - - stop-bit encoded length of the value + - [possible alignment] + - encoded length of the value - bytes for the value. */ static final int LOCK_OFFSET = 0; // 64-bit @@ -685,10 +658,10 @@ class VanillaChronicleMap extends AbstractMap return bytes.startAddr() + offset; } - private long entrySize(long keyLen, long valueLen) { + private long entrySize(long keySize, long valueSize) { return alignment.alignAddr(metaDataBytes + - expectedStopBits(keyLen) + keyLen + - expectedStopBits(valueLen)) + valueLen; + keySizeMarshaller.sizeEncodingSize(keySize) + keySize + + valueSizeMarshaller.sizeEncodingSize(valueSize)) + valueSize; } int inBlocks(long sizeInBytes) { @@ -709,27 +682,30 @@ class VanillaChronicleMap extends AbstractMap * true}, {@code usingValue} or a newly created instance of value class, if {@code usingValue == * null}, is put into this Segment for the key. * - * @param keyBytes serialized {@code key} * @param hash2 a hash code related to the {@code keyBytes} * @return the value which is finally associated with the given key in this Segment after execution of * this method, or {@code null}. */ - V acquire(Bytes keyBytes, K key, V usingValue, int hash2, boolean create) { + V acquire(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue, + int hash2, boolean create) { lock(); try { + long keySize = metaKeyInterop.size(keyInterop, key); MultiStoreBytes entry = tmpBytes; - long offset = searchKey(keyBytes, hash2, entry, hashLookup); + long offset = searchKey(keyInterop, metaKeyInterop, key, keySize, hash2, entry, + hashLookup); if (offset >= 0) { - return onKeyPresentOnAcquire(key, usingValue, offset, entry); + return onKeyPresentOnAcquire(copies, key, usingValue, offset, entry); } else { boolean usingValuePassed = usingValue != null; - usingValue = tryObtainUsingValueOnAcquire(keyBytes, key, usingValue, create); + usingValue = tryObtainUsingValueOnAcquire(key, usingValue, create); if (usingValue != null) { // If `create` is false, this method was called from get() or getUsing() // and non-null `usingValue` was returned by notifyMissed() method. // This "missed" default value is considered as genuine value // rather than "using" container to fill up, even if it implements Byteable. - offset = putEntry(keyBytes, usingValue, create); + offset = putEntry(copies, metaKeyInterop, keyInterop, key, keySize, + usingValue, create); incrementSize(); if (usingValuePassed || !create) notifyPut(offset, true, key, usingValue, posFromOffset(offset)); @@ -743,30 +719,30 @@ class VanillaChronicleMap extends AbstractMap } } - long searchKey(Bytes keyBytes, int hash2, + long searchKey(KI keyInterop, MKI metaKeyInterop, K key, long keySize, int hash2, MultiStoreBytes entry, IntIntMultiMap hashLookup) { - long keyLen = keyBytes.remaining(); hashLookup.startSearch(hash2); for (int pos; (pos = hashLookup.nextPos()) >= 0; ) { long offset = offsetFromPos(pos); reuse(entry, offset); - if (!keyEquals(keyBytes, keyLen, entry)) + if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) continue; // key is found - entry.skip(keyLen); + entry.skip(keySize); return offset; } // key is not found return -1L; } - V onKeyPresentOnAcquire(K key, V usingValue, long offset, NativeBytes entry) { - V v = readValue(entry, usingValue); + V onKeyPresentOnAcquire(ThreadLocalCopies copies, K key, V usingValue, long offset, + NativeBytes entry) { + V v = readValue(copies, entry, usingValue); notifyGet(offset, key, v); return v; } - V tryObtainUsingValueOnAcquire(Bytes keyBytes, K key, V usingValue, boolean create) { + V tryObtainUsingValueOnAcquire(K key, V usingValue, boolean create) { if (create) { if (usingValue != null) { return usingValue; @@ -789,30 +765,33 @@ class VanillaChronicleMap extends AbstractMap } else { if (usingValue instanceof Byteable) ((Byteable) usingValue).bytes(null, 0); - return notifyMissed(keyBytes, key, usingValue); + return notifyMissed(key, usingValue); } } - V put(Bytes keyBytes, K key, V value, int hash2, boolean replaceIfPresent) { + V put(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V value, + int hash2, boolean replaceIfPresent) { lock(); try { - long keyLen = keyBytes.remaining(); + long keySize = metaKeyInterop.size(keyInterop, key); hashLookup.startSearch(hash2); for (int pos; (pos = hashLookup.nextPos()) >= 0; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); - if (!keyEquals(keyBytes, keyLen, entry)) + if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) continue; // key is found - entry.skip(keyLen); + entry.skip(keySize); if (replaceIfPresent) { - return replaceValueOnPut(key, value, entry, pos, offset, !putReturnsNull, hashLookup); + return replaceValueOnPut(copies, key, value, entry, pos, offset, + !putReturnsNull, hashLookup); } else { - return putReturnsNull ? null : readValue(entry, null); + return putReturnsNull ? null : readValue(copies, entry, null); } } // key is not found - long offset = putEntry(keyBytes, value, false); + long offset = putEntry(copies, metaKeyInterop, keyInterop, key, keySize, value, + false); incrementSize(); notifyPut(offset, true, key, value, posFromOffset(offset)); return null; @@ -821,74 +800,81 @@ class VanillaChronicleMap extends AbstractMap } } - V replaceValueOnPut(K key, V value, NativeBytes entry, int pos, long offset, - boolean readPrevValue, IntIntMultiMap searchedHashLookup) { - long valueLenPos = entry.position(); - long valueLen = readValueLen(entry); - long entryEndAddr = entry.positionAddr() + valueLen; + V replaceValueOnPut(ThreadLocalCopies copies, K key, V value, NativeBytes entry, int pos, + long offset, boolean readPrevValue, IntIntMultiMap searchedHashLookup) { + long valueSizePos = entry.position(); + long valueSize = readValueSize(entry); + long entryEndAddr = entry.positionAddr() + valueSize; V prevValue = null; if (readPrevValue) - prevValue = readValue(entry, null, valueLen); + prevValue = readValue(copies, entry, null, valueSize); // putValue may relocate entry and change offset - offset = putValue(pos, offset, entry, valueLenPos, entryEndAddr, value, searchedHashLookup); + offset = putValue(pos, offset, entry, valueSizePos, entryEndAddr, copies, value, null, + searchedHashLookup); notifyPut(offset, false, key, value, posFromOffset(offset)); return prevValue; } /** - * Puts entry. If {@code value} implements {@link net.openhft.lang.model.Byteable} interface and + * Puts entry. If {@code value} implements {@link Byteable} interface and * {@code usingValue} is {@code true}, the value is backed with the bytes of this entry. * - * @param keyBytes serialized key * @param value the value to put - * @param usingValue {@code true} if the value should be backed with the bytes of the entry, if it - * implements {@link net.openhft.lang.model.Byteable} interface, {@code false} if it + * @param usingValue {@code true} if the value should be backed with the bytes of the entry, + * if it implements {@link Byteable} interface, {@code false} if it * should put itself * @return offset of the written entry in the Segment bytes */ - private long putEntry(Bytes keyBytes, V value, boolean usingValue) { - long keyLen = keyBytes.remaining(); + private long putEntry(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, + long keySize, V value, boolean usingValue) { // "if-else polymorphism" is not very beautiful, but allows to // reuse the rest code of this method and doesn't hurt performance. boolean byteableValue = usingValue && value instanceof Byteable; - long valueLen; - Bytes valueBytes = null; + long valueSize; + MetaBytesWriter metaValueWriter = null; + VW valueWriter = null; Byteable valueAsByteable = null; if (!byteableValue) { - valueBytes = getValueAsBytes(value); - valueLen = valueBytes.remaining(); + copies = valueWriterProvider.getCopies(copies); + valueWriter = valueWriterProvider.get(copies, originalValueWriter); + copies = metaValueWriterProvider.getCopies(copies); + metaValueWriter = metaValueWriterProvider.get( + copies, originalMetaValueWriter, valueWriter, value); + valueSize = metaValueWriter.size(valueWriter, value); } else { valueAsByteable = (Byteable) value; - valueLen = valueAsByteable.maxSize(); + valueSize = valueAsByteable.maxSize(); } - long entrySize = entrySize(keyLen, valueLen); + long entrySize = entrySize(keySize, valueSize); int pos = alloc(inBlocks(entrySize)); long offset = offsetFromPos(pos); clearMetaData(offset); NativeBytes entry = entry(offset); - entry.writeStopBit(keyLen); - entry.write(keyBytes); + keySizeMarshaller.writeSize(entry, keySize); + metaKeyInterop.write(keyInterop, entry, key); - writeValueOnPutEntry(valueLen, valueBytes, valueAsByteable, entry); + writeValueOnPutEntry(valueSize, metaValueWriter, valueWriter, value, valueAsByteable, + entry); hashLookup.putAfterFailedSearch(pos); return offset; } - void writeValueOnPutEntry(long valueLen, @Nullable Bytes valueBytes, - @Nullable Byteable valueAsByteable, NativeBytes entry) { - entry.writeStopBit(valueLen); + void writeValueOnPutEntry(long valueSize, MetaBytesWriter metaValueWriter, + VW valueWriter, V value, Byteable valueAsByteable, + NativeBytes entry) { + valueSizeMarshaller.writeSize(entry, valueSize); alignment.alignPositionAddr(entry); - if (valueBytes != null) { - entry.write(valueBytes); + if (metaValueWriter != null) { + metaValueWriter.write(valueWriter, entry, value); } else { assert valueAsByteable != null; long valueOffset = entry.positionAddr() - bytes.address(); - bytes.zeroOut(valueOffset, valueOffset + valueLen); + bytes.zeroOut(valueOffset, valueOffset + valueSize); valueAsByteable.bytes(bytes, valueOffset); } } @@ -938,55 +924,48 @@ class VanillaChronicleMap extends AbstractMap nextPosToSearchFrom = fromPos; } - V readValue(NativeBytes entry, V value) { - return readValue(entry, value, readValueLen(entry)); + V readValue(ThreadLocalCopies copies, NativeBytes entry, V value) { + return readValue(copies, entry, value, readValueSize(entry)); } - long readValueLen(Bytes entry) { - long valueLen = entry.readStopBit(); - alignment.alignPositionAddr(entry); - return valueLen; - } - - /** - * TODO use the value length to limit reading - * - * @param value the object to reuse (if possible), if {@code null} a new object is created - */ - V readValue(NativeBytes entry, V value, long valueLen) { + V readValue(ThreadLocalCopies copies, NativeBytes entry, V value, long valueSize) { + copies = valueReaderProvider.getCopies(copies); + BytesReader valueReader = valueReaderProvider.get(copies, originalValueReader); bytes.positionAddr(entry.positionAddr()); - return valueMarshaller.read(bytes, value); + return valueReader.read(bytes, valueSize, value); } - boolean keyEquals(Bytes keyBytes, long keyLen, Bytes entry) { - return keyLen == entry.readStopBit() && entry.startsWith(keyBytes); + boolean keyEquals(KI keyInterop, MKI metaKeyInterop, K key, + long keySize, Bytes entry) { + return keySize == keySizeMarshaller.readSize(entry) && + metaKeyInterop.startsWith(keyInterop, entry, key); } /** * Removes a key (or key-value pair) from the Segment.

The entry will only be removed if {@code * expectedValue} equals to {@code null} or the value previously corresponding to the specified key. * - * @param keyBytes bytes of the key to remove * @param hash2 a hash code related to the {@code keyBytes} * @return the value of the entry that was removed if the entry corresponding to the {@code keyBytes} * exists and {@link #removeReturnsNull} is {@code false}, {@code null} otherwise */ - V remove(Bytes keyBytes, K key, V expectedValue, int hash2) { + V remove(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, + V expectedValue, int hash2) { lock(); try { - long keyLen = keyBytes.remaining(); + long keySize = metaKeyInterop.size(keyInterop, key); hashLookup.startSearch(hash2); for (int pos; (pos = hashLookup.nextPos()) >= 0; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); - if (!keyEquals(keyBytes, keyLen, entry)) + if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) continue; // key is found - entry.skip(keyLen); - long valueLen = readValueLen(entry); - long entryEndAddr = entry.positionAddr() + valueLen; + entry.skip(keySize); + long valueSize = readValueSize(entry); + long entryEndAddr = entry.positionAddr() + valueSize; V valueRemoved = expectedValue != null || !removeReturnsNull - ? readValue(entry, null, valueLen) : null; + ? readValue(copies, entry, null, valueSize) : null; if (expectedValue != null && !expectedValue.equals(valueRemoved)) return null; hashLookup.removePrevPos(); @@ -1002,15 +981,15 @@ class VanillaChronicleMap extends AbstractMap } } - boolean containsKey(Bytes keyBytes, int hash2) { + boolean containsKey(KI keyInterop, MKI metaKeyInterop, K key, int hash2) { lock(); try { - long keyLen = keyBytes.remaining(); + long keySize = metaKeyInterop.size(keyInterop, key); IntIntMultiMap hashLookup = containsKeyHashLookup(); hashLookup.startSearch(hash2); for (int pos; (pos = hashLookup.nextPos()) >= 0; ) { Bytes entry = entry(offsetFromPos(pos)); - if (keyEquals(keyBytes, keyLen, entry)) + if (keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) return true; } return false; @@ -1031,20 +1010,21 @@ class VanillaChronicleMap extends AbstractMap * @param hash2 a hash code related to the {@code keyBytes} * @return the replaced value or {@code null} if the value was not replaced */ - V replace(Bytes keyBytes, K key, V expectedValue, V newValue, int hash2) { + V replace(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, + V expectedValue, V newValue, int hash2) { lock(); try { - long keyLen = keyBytes.remaining(); + long keySize = metaKeyInterop.size(keyInterop, key); hashLookup.startSearch(hash2); for (int pos; (pos = hashLookup.nextPos()) >= 0; ) { long offset = offsetFromPos(pos); NativeBytes entry = entry(offset); - if (!keyEquals(keyBytes, keyLen, entry)) + if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) continue; // key is found - entry.skip(keyLen); - return onKeyPresentOnReplace(key, expectedValue, newValue, pos, offset, entry, - hashLookup); + entry.skip(keySize); + return onKeyPresentOnReplace(copies, key, expectedValue, newValue, + pos, offset, entry, hashLookup); } // key is not found return null; @@ -1053,18 +1033,19 @@ class VanillaChronicleMap extends AbstractMap } } - V onKeyPresentOnReplace(K key, V expectedValue, V newValue, int pos, long offset, - NativeBytes entry, IntIntMultiMap searchedHashLookup) { - long valueLenPos = entry.position(); - long valueLen = readValueLen(entry); - long entryEndAddr = entry.positionAddr() + valueLen; - V valueRead = readValue(entry, null, valueLen); + V onKeyPresentOnReplace(ThreadLocalCopies copies, K key, V expectedValue, V newValue, + int pos, long offset, NativeBytes entry, + IntIntMultiMap searchedHashLookup) { + long valueSizePos = entry.position(); + long valueSize = readValueSize(entry); + long entryEndAddr = entry.positionAddr() + valueSize; + V valueRead = readValue(copies, entry, null, valueSize); if (valueRead == null) return null; if (expectedValue == null || expectedValue.equals(valueRead)) { // putValue may relocate entry and change offset - offset = putValue(pos, offset, entry, valueLenPos, entryEndAddr, newValue, - searchedHashLookup); + offset = putValue(pos, offset, entry, valueSizePos, entryEndAddr, copies, newValue, + null, searchedHashLookup); notifyPut(offset, false, key, newValue, posFromOffset(offset)); return valueRead; @@ -1089,10 +1070,9 @@ class VanillaChronicleMap extends AbstractMap } } - V notifyMissed(Bytes keyBytes, K key, V usingValue) { + V notifyMissed(K key, V usingValue) { if (eventListener() != MapEventListeners.NOP) { - return eventListener().onGetMissing(VanillaChronicleMap.this, keyBytes, - key, usingValue); + return eventListener().onGetMissing(VanillaChronicleMap.this, key, usingValue); } return null; } @@ -1105,17 +1085,6 @@ class VanillaChronicleMap extends AbstractMap } } - long putValue(int pos, long offset, NativeBytes entry, long valueLenPos, - long entryEndAddr, V value, IntIntMultiMap searchedHashLookup) { - if (value instanceof Byteable) { - return putValue(pos, offset, entry, valueLenPos, entryEndAddr, - null, (Byteable) value, false, searchedHashLookup); - } else { - return putValue(pos, offset, entry, valueLenPos, entryEndAddr, - getValueAsBytes(value), null, true, searchedHashLookup); - } - } - /** * Replaces value in existing entry. May cause entry relocation, because there may be not enough space * for new value in location already allocated for this entry. @@ -1124,30 +1093,31 @@ class VanillaChronicleMap extends AbstractMap * @param offset relative offset of the entry in Segment bytes (before, i. e. including * metaData) * @param entry relative pointer in Segment bytes - * @param valueLenPos relative position of value "stop bit" in entry + * @param valueSizePos relative position of value size in entry * @param entryEndAddr absolute address of the entry end - * @param valueBytes serialized value, or {@code null} if valueAsByteable is given - * @param valueAsByteable the value to put as {@code Byteable}, or {@code null} if valueBytes is - * given - * @param allowOversize {@code true} if the entry is allowed become oversized if it was not yet * @return relative offset of the entry in Segment bytes after putting value (that may cause entry * relocation) */ - long putValue(int pos, long offset, NativeBytes entry, long valueLenPos, - long entryEndAddr, @Nullable Bytes valueBytes, - @Nullable Byteable valueAsByteable, boolean allowOversize, + long putValue(int pos, long offset, NativeBytes entry, long valueSizePos, + long entryEndAddr, ThreadLocalCopies copies, V value, Bytes valueBytes, IntIntMultiMap searchedHashLookup) { - long valueLenAddr = entry.address() + valueLenPos; - long newValueLen; + long valueSizeAddr = entry.address() + valueSizePos; + long newValueSize; + VW valueWriter = null; + MetaBytesWriter metaValueWriter = null; if (valueBytes != null) { - newValueLen = valueBytes.remaining(); + newValueSize = valueBytes.remaining(); } else { - assert valueAsByteable != null; - newValueLen = valueAsByteable.maxSize(); + copies = valueWriterProvider.getCopies(copies); + valueWriter = valueWriterProvider.get(copies, originalValueWriter); + copies = metaValueWriterProvider.getCopies(copies); + metaValueWriter = metaValueWriterProvider.get( + copies, originalMetaValueWriter, valueWriter, value); + newValueSize = metaValueWriter.size(valueWriter, value); } long newValueAddr = alignment.alignAddr( - valueLenAddr + expectedStopBits(newValueLen)); - long newEntryEndAddr = newValueAddr + newValueLen; + valueSizeAddr + valueSizeMarshaller.sizeEncodingSize(newValueSize)); + long newEntryEndAddr = newValueAddr + newValueSize; // Fast check before counting "sizes in blocks" that include // integral division newValueDoesNotFit: @@ -1157,24 +1127,7 @@ class VanillaChronicleMap extends AbstractMap int oldSizeInBlocks = inBlocks(oldEntrySize); int newSizeInBlocks = inBlocks(newEntryEndAddr - entryStartAddr); if (newSizeInBlocks > oldSizeInBlocks) { - if (!allowOversize && oldSizeInBlocks == 1) { - // If the value is Byteable, illegal oversize could be detected because of - // too high (inaccurate maxSize() estimate. Marshall the value to get - // the precise size. - if (valueAsByteable != null) { - // noinspection unchecked - return putValue(pos, offset, entry, valueLenPos, entryEndAddr, - getValueAsBytes((V) valueAsByteable), null, false, searchedHashLookup); - } - throw new IllegalArgumentException("Byteable value is not allowed " + - "to make the entry oversized while it was not initially."); - } if (newSizeInBlocks > MAX_ENTRY_OVERSIZE_FACTOR) { - if (valueAsByteable != null) { - // noinspection unchecked - return putValue(pos, offset, entry, valueLenPos, entryEndAddr, - getValueAsBytes((V) valueAsByteable), null, false, searchedHashLookup); - } throw new IllegalArgumentException("Value too large: " + "entry takes " + newSizeInBlocks + " blocks, " + MAX_ENTRY_OVERSIZE_FACTOR + " is maximum."); @@ -1190,12 +1143,12 @@ class VanillaChronicleMap extends AbstractMap // after successful search by key replacePosInHashLookupOnRelocation(searchedHashLookup, prevPos, pos); offset = offsetFromPos(pos); - // Moving metadata, key stop bit and key. + // Moving metadata, key size and key. // Don't want to fiddle with pseudo-buffers for this, // since we already have all absolute addresses. long newEntryStartAddr = entryStartAddr(offset); NativeBytes.UNSAFE.copyMemory(entryStartAddr, - newEntryStartAddr, valueLenAddr - entryStartAddr); + newEntryStartAddr, valueSizeAddr - entryStartAddr); entry = entry(offset); // END OF RELOCATION } else if (newSizeInBlocks < oldSizeInBlocks) { @@ -1208,23 +1161,13 @@ class VanillaChronicleMap extends AbstractMap } } // Common code for all cases - entry.position(valueLenPos); - entry.writeStopBit(newValueLen); + entry.position(valueSizePos); + valueSizeMarshaller.writeSize(entry, newValueSize); alignment.alignPositionAddr(entry); if (valueBytes != null) { entry.write(valueBytes); } else { - if (valueAsByteable instanceof BytesMarshallable) { - long posAddr = entry.positionAddr(); - ((BytesMarshallable) valueAsByteable).writeMarshallable(entry); - long actualValueLen = entry.positionAddr() - posAddr; - if (actualValueLen > newValueLen) { - throw new AssertionError( - "Byteable value returned maxSize less than the actual size"); - } - } else { - entry.write(valueAsByteable.bytes(), valueAsByteable.offset(), newValueLen); - } + metaValueWriter.write(valueWriter, entry, value); } return offset; } @@ -1246,11 +1189,16 @@ class VanillaChronicleMap extends AbstractMap public Entry getEntry(long pos) { bytes.position(offsetFromPos(pos) + metaDataBytes); - bytes.readStopBit(); - K key = keyMarshaller.read(bytes); - bytes.readStopBit(); + + long keySize = keySizeMarshaller.readSize(bytes); + ThreadLocalCopies copies = keyReaderProvider.getCopies(null); + K key = keyReaderProvider.get(copies, originalKeyReader).read(bytes, keySize); + + long valueSize = valueSizeMarshaller.readSize(bytes); alignment.alignPositionAddr(bytes); - V value = valueMarshaller.read(bytes); + copies = valueReaderProvider.getCopies(copies); + V value = valueReaderProvider.get(copies, originalValueReader).read(bytes, valueSize); + return new WriteThroughEntry(key, value); } @@ -1268,11 +1216,11 @@ class VanillaChronicleMap extends AbstractMap throw new AssertionError(); long offset = offsetFromPos(pos); Bytes entry = entry(offset); - long keyLen = entry.readStopBit(); - entry.skip(keyLen); + long keySize = keySizeMarshaller.readSize(entry); + entry.skip(keySize); afterKeyHookOnCheckConsistency(entry); - long valueLen = entry.readStopBit(); - long sizeInBytes = entrySize(keyLen, valueLen); + long valueSize = valueSizeMarshaller.readSize(entry); + long sizeInBytes = entrySize(keySize, valueSize); int entrySizeInBlocks = inBlocks(sizeInBytes); if (!freeList.allSet(pos, pos + entrySizeInBlocks)) throw new AssertionError(); @@ -1400,14 +1348,13 @@ class VanillaChronicleMap extends AbstractMap final NativeBytes entry = segment.entry(offset); final long limit = entry.limit(); - final long keyLen = entry.readStopBit(); - entry.limit(entry.position() + keyLen); - final int segmentHash = hasher.segmentHash(hash(entry)); - entry.limit(limit); + final long keySize = keySizeMarshaller.readSize(entry); + long position = entry.position(); + final int segmentHash = segmentHash(Hasher.hash(entry, position, position + keySize)); - entry.skip(keyLen); - long valueLen = segment.readValueLen(entry); - final long entryEndAddr = entry.positionAddr() + valueLen; + entry.skip(keySize); + long valueSize = readValueSize(entry); + final long entryEndAddr = entry.positionAddr() + valueSize; segment.getHashLookup().remove(segmentHash, pos); segment.decrementSize(); segment.free(pos, segment.inBlocks(entryEndAddr - segment.entryStartAddr(offset))); diff --git a/src/main/java/net/openhft/chronicle/map/serialization/AgileBytesMarshaller.java b/src/main/java/net/openhft/chronicle/map/serialization/AgileBytesMarshaller.java new file mode 100644 index 000000000..d3cd897b1 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/AgileBytesMarshaller.java @@ -0,0 +1,20 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +public interface AgileBytesMarshaller extends BytesReader, BytesInterop, SizeMarshaller { +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/BytesInterop.java b/src/main/java/net/openhft/chronicle/map/serialization/BytesInterop.java new file mode 100644 index 000000000..71a23a127 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/BytesInterop.java @@ -0,0 +1,35 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; +import net.openhft.lang.model.Byteable; + +/** + * Marshaller for byte sequences, which are copied to off-heap {@link Bytes} + * in a very straightforward manner, e. g. {@link Byteable Byteables}, {@code byte[]} arrays, + * {@code Bytes} themselves. The criterion of this interface applicability -- + * {@link #equals} implementation shouldn't require deserialization and any garbage creation. + * + * @param type of marshalled objects + */ +public interface BytesInterop extends BytesWriter { + + boolean startsWith(Bytes bytes, E e); + + long hash(E e); +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/BytesReader.java b/src/main/java/net/openhft/chronicle/map/serialization/BytesReader.java new file mode 100644 index 000000000..e3dcabe48 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/BytesReader.java @@ -0,0 +1,28 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; + +import java.io.Serializable; + +public interface BytesReader extends Serializable { + + E read(Bytes bytes, long size); + + E read(Bytes bytes, long size, E e); +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/BytesReaders.java b/src/main/java/net/openhft/chronicle/map/serialization/BytesReaders.java new file mode 100644 index 000000000..8f9fa04eb --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/BytesReaders.java @@ -0,0 +1,48 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.BytesMarshaller; + +public final class BytesReaders { + + public static BytesReader fromBytesMarshaller(BytesMarshaller marshaller) { + return new SimpleBytesReader(marshaller); + } + + private static class SimpleBytesReader implements BytesReader { + + private final BytesMarshaller marshaller; + + public SimpleBytesReader(BytesMarshaller marshaller) { + this.marshaller = marshaller; + } + + @Override + public E read(Bytes bytes, long size) { + return marshaller.read(bytes); + } + + @Override + public E read(Bytes bytes, long size, E e) { + return marshaller.read(bytes, e); + } + } + + private BytesReaders() {} +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/BytesWriter.java b/src/main/java/net/openhft/chronicle/map/serialization/BytesWriter.java new file mode 100644 index 000000000..ed8ecad11 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/BytesWriter.java @@ -0,0 +1,28 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; + +import java.io.Serializable; + +public interface BytesWriter extends Serializable { + + long size(E e); + + void write(Bytes bytes, E e); +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/CopyingMetaBytesInterop.java b/src/main/java/net/openhft/chronicle/map/serialization/CopyingMetaBytesInterop.java new file mode 100644 index 000000000..1c1922a8f --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/CopyingMetaBytesInterop.java @@ -0,0 +1,242 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.DirectBytes; +import net.openhft.lang.io.DirectStore; +import net.openhft.lang.io.serialization.BytesMarshaller; +import net.openhft.lang.io.serialization.JDKObjectSerializer; +import net.openhft.chronicle.map.threadlocal.Provider; +import net.openhft.chronicle.map.threadlocal.StatefulCopyable; +import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; + +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.Serializable; + +public abstract class CopyingMetaBytesInterop implements MetaBytesInterop { + private static final long serialVersionUID = 0L; + + final DirectBytesBuffer buffer; + transient long size; + transient W writer; + transient E cur; + + private CopyingMetaBytesInterop(DirectBytesBuffer buffer) { + this.buffer = buffer; + } + + void init(W writer, E e, boolean mutable, long maxSize) { + if (mutable || writer != this.writer || e != cur) { + this.writer = writer; + cur = e; + Bytes buffer = this.buffer.obtain(maxSize); + innerWrite(writer, buffer, e); + buffer.flip(); + size = buffer.remaining(); + } + } + + abstract void innerWrite(W writer, Bytes bytes, E e); + + @Override + public long size(W writer, E e) { + return size; + } + + @Override + public boolean startsWith(W writer, Bytes bytes, E e) { + return bytes.startsWith(buffer.buffer); + } + + @Override + public long hash(W writer, E e) { + return Hasher.hash(buffer.buffer); + } + + @Override + public void write(W writer, Bytes bytes, E e) { + bytes.write(buffer.buffer); + } + + DirectBytesBuffer buffer() { + return buffer; + } + + + public static > + MetaBytesInterop forBytesMarshaller(Serializable bufferIdentity) { + return new DirectBytesBuffer(bufferIdentity).forBytesMarshaller; + } + + public static > + MetaBytesInterop forBytesWriter(Serializable bufferIdentity) { + return new DirectBytesBuffer(bufferIdentity).forBytesWriter; + } + + private static final Provider provider = + Provider.of(DirectBytesBuffer.class); + + private static abstract class AbstractCopyingMetaBytesInteropProvider> + implements MetaProvider { + private static final long serialVersionUID = 0L; + + @Override + public ThreadLocalCopies getCopies(ThreadLocalCopies copies) { + return provider.getCopies(copies); + } + } + + public static > + MetaProvider> providerForBytesMarshaller(boolean mutable, + long maxSize) { + return new BytesMarshallerCopyingMetaBytesInteropProvider(mutable, maxSize); + } + + private static class BytesMarshallerCopyingMetaBytesInteropProvider> + extends AbstractCopyingMetaBytesInteropProvider> { + private static final long serialVersionUID = 0L; + private final boolean mutable; + private final long maxSize; + + public BytesMarshallerCopyingMetaBytesInteropProvider(boolean mutable, long maxSize) { + this.mutable = mutable; + this.maxSize = maxSize; + } + + @Override + public CopyingMetaBytesInterop get( + ThreadLocalCopies copies, + CopyingMetaBytesInterop originalMetaWriter, M writer, E e) { + DirectBytesBuffer.ForBytesMarshaller forBytesMarshaller = + provider.get(copies, originalMetaWriter.buffer()).forBytesMarshaller; + forBytesMarshaller.init(writer, e, mutable, maxSize); + return forBytesMarshaller; + } + } + + public static > + MetaProvider> providerForBytesWriter(boolean mutable, + long maxSize) { + return new BytesWriterCopyingMetaBytesInteropProvider(mutable, maxSize); + } + + private static class BytesWriterCopyingMetaBytesInteropProvider> + extends AbstractCopyingMetaBytesInteropProvider> { + private static final long serialVersionUID = 0L; + private final boolean mutable; + private final long maxSize; + + public BytesWriterCopyingMetaBytesInteropProvider(boolean mutable, long maxSize) { + this.mutable = mutable; + this.maxSize = maxSize; + } + + @Override + public CopyingMetaBytesInterop get( + ThreadLocalCopies copies, + CopyingMetaBytesInterop originalMetaWriter, W writer, E e) { + DirectBytesBuffer.ForBytesWriter forBytesWriter = + provider.get(copies, originalMetaWriter.buffer()).forBytesWriter; + forBytesWriter.init(writer, e, mutable, maxSize); + return forBytesWriter; + } + } + + + private static class DirectBytesBuffer + implements StatefulCopyable, Serializable { + private static final long serialVersionUID = 0L; + private final Serializable identity; + private transient DirectBytes buffer; + + private transient ForBytesMarshaller forBytesMarshaller; + private transient ForBytesWriter forBytesWriter; + + DirectBytesBuffer(Serializable identity) { + this.identity = identity; + initTransients(); + } + + private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { + in.defaultReadObject(); + initTransients(); + } + + private void initTransients() { + forBytesMarshaller = new ForBytesMarshaller(); + forBytesWriter = new ForBytesWriter(); + } + + private Bytes obtain(long maxSize) { + DirectBytes buf; + if ((buf = buffer) != null) { + if (maxSize <= buf.capacity()) { + return buf.clear(); + } else { + DirectStore store = (DirectStore) buf.store(); + store.resize(maxSize, false); + return buffer = store.bytes(); + } + } else { + buffer = new DirectStore(JDKObjectSerializer.INSTANCE, maxSize, false).bytes(); + return buffer; + } + } + + @Override + public Object stateIdentity() { + return identity; + } + + @Override + public DirectBytesBuffer copy() { + return new DirectBytesBuffer(identity); + } + + private class ForBytesMarshaller> + extends CopyingMetaBytesInterop { + private static final long serialVersionUID = 0L; + + private ForBytesMarshaller() { + super(DirectBytesBuffer.this); + } + + @Override + void innerWrite(M writer, Bytes bytes, E e) { + writer.write(bytes, e); + } + } + + private class ForBytesWriter> + extends CopyingMetaBytesInterop { + private static final long serialVersionUID = 0L; + + private ForBytesWriter() { + super(DirectBytesBuffer.this); + } + + @Override + void innerWrite(W writer, Bytes bytes, E e) { + writer.write(bytes, e); + } + } + } +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/DelegatingMetaBytesInterop.java b/src/main/java/net/openhft/chronicle/map/serialization/DelegatingMetaBytesInterop.java new file mode 100644 index 000000000..4413252a9 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/DelegatingMetaBytesInterop.java @@ -0,0 +1,58 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; + +import java.io.ObjectStreamException; + +public final class DelegatingMetaBytesInterop> + implements MetaBytesInterop { + private static final long serialVersionUID = 0L; + private static final DelegatingMetaBytesInterop INSTANCE = new DelegatingMetaBytesInterop(); + + public static > + DelegatingMetaBytesInterop instance() { + return INSTANCE; + } + + private DelegatingMetaBytesInterop() {} + + @Override + public long size(I interop, E e) { + return interop.size(e); + } + + @Override + public boolean startsWith(I interop, Bytes bytes, E e) { + return interop.startsWith(bytes, e); + } + + @Override + public long hash(I interop, E e) { + return interop.hash(e); + } + + @Override + public void write(I interop, Bytes bytes, E e) { + interop.write(bytes, e); + } + + private Object readResolve() throws ObjectStreamException { + return INSTANCE; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/DelegatingMetaBytesInteropProvider.java b/src/main/java/net/openhft/chronicle/map/serialization/DelegatingMetaBytesInteropProvider.java new file mode 100644 index 000000000..2544d2d61 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/DelegatingMetaBytesInteropProvider.java @@ -0,0 +1,50 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; + +import java.io.ObjectStreamException; + +public final class DelegatingMetaBytesInteropProvider> + implements MetaProvider> { + private static final long serialVersionUID = 0L; + private static final DelegatingMetaBytesInteropProvider INSTANCE = + new DelegatingMetaBytesInteropProvider(); + + public static > + MetaProvider> instance() { + return INSTANCE; + } + + private DelegatingMetaBytesInteropProvider() {} + + @Override + public MetaBytesInterop get(ThreadLocalCopies copies, + MetaBytesInterop originalMetaWriter, I writer, E e) { + return originalMetaWriter; + } + + @Override + public ThreadLocalCopies getCopies(ThreadLocalCopies copies) { + return copies; + } + + private Object readResolve() throws ObjectStreamException { + return INSTANCE; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/DirectBytesBuffer.java b/src/main/java/net/openhft/chronicle/map/serialization/DirectBytesBuffer.java new file mode 100644 index 000000000..1f454a62b --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/DirectBytesBuffer.java @@ -0,0 +1,58 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.DirectBytes; +import net.openhft.lang.io.DirectStore; +import net.openhft.lang.io.serialization.JDKObjectSerializer; +import net.openhft.chronicle.map.threadlocal.StatefulCopyable; + +final class DirectBytesBuffer implements StatefulCopyable { + + private final Object identity; + private DirectBytes buffer; + + DirectBytesBuffer(Object identity) { + this.identity = identity; + } + + private Bytes buffer(long maxSize) { + DirectBytes buf; + if ((buf = buffer) != null) { + if (maxSize <= buf.capacity()) { + return buf.clear(); + } else { + DirectStore store = (DirectStore) buf.store(); + store.resize(maxSize, false); + return buffer = store.bytes(); + } + } else { + return buffer = new DirectStore(JDKObjectSerializer.INSTANCE, maxSize, false).bytes(); + } + } + + @Override + public Object stateIdentity() { + return identity; + } + + @Override + public DirectBytesBuffer copy() { + return new DirectBytesBuffer(identity); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/Hasher.java b/src/main/java/net/openhft/chronicle/map/serialization/Hasher.java new file mode 100644 index 000000000..250172810 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/Hasher.java @@ -0,0 +1,62 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.AbstractBytes; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.NativeBytes; + +public final class Hasher { + + public static long hash(Bytes bytes) { + return hash(bytes, bytes.position(), bytes.limit()); + } + + public static long hash(Bytes bytes, long offset, long limit) { + long h = 0; + long i = offset; + for (; i < limit - 7; i += 8) + h = 1011001110001111L * h + bytes.readLong(i); + for (; i < limit - 1; i += 2) + h = 101111 * h + bytes.readShort(i); + if (i < limit) + h = 2111 * h + bytes.readByte(i); + return hash(h); + } + + public static long hash(byte[] ba) { + long h = 0; + long i = 0; + long base = NativeBytes.UNSAFE.arrayBaseOffset(byte[].class); + long limit = ba.length; + for (; i < limit - 7; i += 8) + h = 1011001110001111L * h + NativeBytes.UNSAFE.getLong(ba, base + i); + for (; i < limit - 1; i += 2) + h = 101111 * h + NativeBytes.UNSAFE.getShort(ba, base + i); + if (i < limit) + h = 2111 * h + NativeBytes.UNSAFE.getByte(ba, base + i); + return hash(h); + } + + public static long hash(long value) { + value *= 11018881818881011L; + value ^= (value >>> 41) ^ (value >>> 21); + return value; + } + + private Hasher() {} +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/MetaBytesInterop.java b/src/main/java/net/openhft/chronicle/map/serialization/MetaBytesInterop.java new file mode 100644 index 000000000..d900b9bfd --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/MetaBytesInterop.java @@ -0,0 +1,27 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; + +public interface MetaBytesInterop extends MetaBytesWriter { + + boolean startsWith(I interop, Bytes bytes, E e); + + long hash(I interop, E e); + +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/MetaBytesWriter.java b/src/main/java/net/openhft/chronicle/map/serialization/MetaBytesWriter.java new file mode 100644 index 000000000..493bee0d7 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/MetaBytesWriter.java @@ -0,0 +1,27 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; + +import java.io.Serializable; + +public interface MetaBytesWriter extends Serializable { + long size(W writer, E e); + + void write(W writer, Bytes bytes, E e); +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/MetaProvider.java b/src/main/java/net/openhft/chronicle/map/serialization/MetaProvider.java new file mode 100644 index 000000000..b9620ce61 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/MetaProvider.java @@ -0,0 +1,28 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; + +import java.io.Serializable; + +public interface MetaProvider> extends Serializable { + + MW get(ThreadLocalCopies copies, MW originalMetaWriter, W writer, E e); + + ThreadLocalCopies getCopies(ThreadLocalCopies copies); +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/SizeMarshaller.java b/src/main/java/net/openhft/chronicle/map/serialization/SizeMarshaller.java new file mode 100644 index 000000000..024af8558 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/SizeMarshaller.java @@ -0,0 +1,30 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; + +import java.io.Serializable; + +public interface SizeMarshaller extends Serializable { + + int sizeEncodingSize(long size); + + void writeSize(Bytes bytes, long size); + + long readSize(Bytes bytes); +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/SizeMarshallers.java b/src/main/java/net/openhft/chronicle/map/serialization/SizeMarshallers.java new file mode 100644 index 000000000..6a26114b7 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/SizeMarshallers.java @@ -0,0 +1,51 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization; + +import net.openhft.lang.io.Bytes; + +public final class SizeMarshallers { + + public static SizeMarshaller stopBit() { + return StopBit.INSTANCE; + } + + private enum StopBit implements SizeMarshaller { + INSTANCE; + + @Override + public int sizeEncodingSize(long size) { + if (size <= 127) + return 1; + // numberOfLeadingZeros is cheap intrinsic on modern CPUs + // integral division is not... but there is no choice + return ((70 - Long.numberOfLeadingZeros(size)) / 7); + } + + @Override + public void writeSize(Bytes bytes, long size) { + bytes.writeStopBit(size); + } + + @Override + public long readSize(Bytes bytes) { + return bytes.readStopBit(); + } + } + + private SizeMarshallers() {} +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/impl/ByteArrayMarshaller.java b/src/main/java/net/openhft/chronicle/map/serialization/impl/ByteArrayMarshaller.java new file mode 100644 index 000000000..0fa9c4e26 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/impl/ByteArrayMarshaller.java @@ -0,0 +1,66 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization.impl; + +import net.openhft.chronicle.map.serialization.BytesInterop; +import net.openhft.chronicle.map.serialization.BytesReader; +import net.openhft.chronicle.map.serialization.Hasher; +import net.openhft.lang.io.Bytes; + +public enum ByteArrayMarshaller implements BytesInterop, BytesReader { + INSTANCE; + + @Override + public long size(byte[] ba) { + return ba.length; + } + + @Override + public boolean startsWith(Bytes bytes, byte[] ba) { + long pos = bytes.position(); + for (int i = 0; i < ba.length; i++) { + if (bytes.readByte(pos + i) != ba[i]) + return false; + } + return true; + } + + @Override + public long hash(byte[] ba) { + return Hasher.hash(ba); + } + + @Override + public void write(Bytes bytes, byte[] ba) { + bytes.write(ba); + } + + @Override + public byte[] read(Bytes bytes, long size) { + byte[] ba = new byte[(int) size]; + bytes.read(ba); + return ba; + } + + @Override + public byte[] read(Bytes bytes, long size, byte[] ba) { + if (size != ba.length) + ba = new byte[(int) size]; + bytes.read(ba); + return ba; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/impl/ByteableMarshaller.java b/src/main/java/net/openhft/chronicle/map/serialization/impl/ByteableMarshaller.java new file mode 100644 index 000000000..1f5efd1f7 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/impl/ByteableMarshaller.java @@ -0,0 +1,147 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization.impl; + +import net.openhft.chronicle.map.serialization.AgileBytesMarshaller; +import net.openhft.chronicle.map.serialization.Hasher; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.NativeBytes; +import net.openhft.lang.io.serialization.*; +import net.openhft.lang.io.serialization.impl.AllocateInstanceObjectFactory; +import net.openhft.lang.model.Byteable; +import org.jetbrains.annotations.NotNull; + +public class ByteableMarshaller implements AgileBytesMarshaller { + private static final long serialVersionUID = 0L; + + public static ByteableMarshaller of(@NotNull Class eClass) { + return new ByteableMarshaller(eClass); + } + + public static ByteableMarshaller of(@NotNull Class eClass, + ObjectFactory factory) { + if (factory instanceof AllocateInstanceObjectFactory) { + Class allocatedClass = ((AllocateInstanceObjectFactory) factory).allocatedClass(); + return new ByteableMarshaller(allocatedClass); + } else { + return new WithCustomFactory(eClass, factory); + } + } + + @NotNull + final Class tClass; + final long size; + + ByteableMarshaller(@NotNull Class tClass) { + this.tClass = tClass; + try { + size = getInstance().maxSize(); + } catch (Exception e) { + throw new IllegalStateException(e); + } + } + + @Override + public long size(E e) { + return size; + } + + @Override + public int sizeEncodingSize(long size) { + return 0; + } + + @Override + public void writeSize(Bytes bytes, long size) { + // do nothing + } + + @Override + public boolean startsWith(Bytes bytes, E e) { + Bytes input = e.bytes(); + long pos = bytes.position(), inputPos = e.offset(); + + int i = 0; + for (; i < size - 7; i += 8) { + if (bytes.readLong(pos + i) != input.readLong(inputPos + i)) + return false; + } + for (; i < size; i++) { + if (bytes.readByte(pos + i) != input.readByte(inputPos + i)) + return false; + } + return true; + } + + @Override + public long hash(E e) { + return Hasher.hash(e.bytes(), e.offset(), e.offset() + size); + } + + @Override + public void write(Bytes bytes, E e) { + bytes.write(e.bytes(), e.offset(), size); + } + + @Override + public long readSize(Bytes bytes) { + return size; + } + + @Override + public E read(Bytes bytes, long size) { + return read(bytes, size, null); + } + + @Override + public E read(Bytes bytes, long size, E e) { + try { + if (e == null) + e = getInstance(); + e.bytes(bytes, bytes.position()); + bytes.skip(size); + return e; + } catch (Exception ex) { + throw new IllegalStateException(ex); + } + } + + @SuppressWarnings("unchecked") + @NotNull + E getInstance() throws Exception { + return (E) NativeBytes.UNSAFE.allocateInstance(tClass); + } + + private static class WithCustomFactory extends ByteableMarshaller { + private static final long serialVersionUID = 0L; + + @NotNull + private final ObjectFactory factory; + + WithCustomFactory(@NotNull Class tClass, + @NotNull ObjectFactory factory) { + super(tClass); + this.factory = factory; + } + + @NotNull + @Override + E getInstance() throws Exception { + return factory.create(); + } + } +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/impl/CharSequenceReader.java b/src/main/java/net/openhft/chronicle/map/serialization/impl/CharSequenceReader.java new file mode 100644 index 000000000..aed9f7c71 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/impl/CharSequenceReader.java @@ -0,0 +1,117 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization.impl; + +import net.openhft.chronicle.map.serialization.BytesReader; +import net.openhft.lang.io.AbstractBytes; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.pool.CharSequenceInterner; +import net.openhft.chronicle.map.threadlocal.StatefulCopyable; +import org.jetbrains.annotations.NotNull; + +import java.io.IOException; +import java.io.ObjectStreamException; +import java.io.Serializable; + +public final class CharSequenceReader + implements BytesReader, StatefulCopyable> { + private static final long serialVersionUID = 0L; + + private enum NoInterning implements CharSequenceInterner { + INSTANCE; + @Override + public String intern(CharSequence cs) { + return cs.toString(); + } + } + + private static final CharSequenceReader DEFAULT_READER = + new CharSequenceReader(NoInterning.INSTANCE, NoInterning.INSTANCE); + + + public static CharSequenceReader of() { + return DEFAULT_READER; + } + + public static & Serializable> + CharSequenceReader of(@NotNull I interner, @NotNull Serializable stateIdentity) { + return new CharSequenceReader(interner, stateIdentity); + } + + private final StringBuilder sb = new StringBuilder(128); + @NotNull + private final CharSequenceInterner interner; + @NotNull + private final Serializable identity; + + private CharSequenceReader(@NotNull CharSequenceInterner interner, + @NotNull Serializable identity) { + this.interner = interner; + this.identity = identity; + } + + @Override + public S read(Bytes bytes, long size) { + sb.setLength(0); + try { + AbstractBytes.readUTF0(bytes, sb, (int) size); + } catch (IOException e) { + throw new IllegalStateException(e); + } + return interner.intern(sb); + } + + @Override + public S read(Bytes bytes, long size, S s) { + Appendable appendable; + if (s instanceof Appendable) { + appendable = (Appendable) s; + if (s instanceof StringBuilder) { + ((StringBuilder) s).setLength(0); + } else if (s instanceof StringBuffer) { + ((StringBuffer) s).setLength(0); + } + } else { + sb.setLength(0); + appendable = sb; + } + try { + AbstractBytes.readUTF0(bytes, appendable, (int) size); + } catch (IOException e) { + throw new IllegalStateException(e); + } + if (appendable == s) + return s; + return interner.intern(sb); + } + + @Override + public Object stateIdentity() { + return identity; + } + + @Override + public CharSequenceReader copy() { + return new CharSequenceReader(interner, identity); + } + + private Object readResolve() throws ObjectStreamException { + if (interner == NoInterning.INSTANCE) + return DEFAULT_READER; + return this; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/impl/CharSequenceWriter.java b/src/main/java/net/openhft/chronicle/map/serialization/impl/CharSequenceWriter.java new file mode 100644 index 000000000..38a5eb6a3 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/impl/CharSequenceWriter.java @@ -0,0 +1,48 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization.impl; + +import net.openhft.lang.io.AbstractBytes; +import net.openhft.lang.io.Bytes; +import net.openhft.chronicle.map.serialization.BytesWriter; + +import java.io.ObjectStreamException; + +public final class CharSequenceWriter implements BytesWriter { + private static final long serialVersionUID = 0L; + private static final CharSequenceWriter INSTANCE = new CharSequenceWriter(); + + public static CharSequenceWriter instance() { + return INSTANCE; + } + + private CharSequenceWriter() {} + + @Override + public long size(CS s) { + return AbstractBytes.findUTFLength(s, s.length()); + } + + @Override + public void write(Bytes bytes, CS s) { + AbstractBytes.writeUTF0(bytes, s, s.length()); + } + + private Object readResolve() throws ObjectStreamException { + return INSTANCE; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/serialization/impl/LongMarshaller.java b/src/main/java/net/openhft/chronicle/map/serialization/impl/LongMarshaller.java new file mode 100644 index 000000000..2506058e9 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/impl/LongMarshaller.java @@ -0,0 +1,70 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.serialization.impl; + +import net.openhft.chronicle.map.serialization.AgileBytesMarshaller; +import net.openhft.chronicle.map.serialization.Hasher; +import net.openhft.lang.io.Bytes; + +public enum LongMarshaller implements AgileBytesMarshaller { + INSTANCE; + + @Override + public long size(Long e) { + return 4L; + } + + @Override + public int sizeEncodingSize(long size) { + return 0; + } + + @Override + public void writeSize(Bytes bytes, long size) { + // do nothing + } + + @Override + public boolean startsWith(Bytes bytes, Long e) { + return e == bytes.readLong(0); + } + + @Override + public long hash(Long e) { + return Hasher.hash(e); + } + + @Override + public void write(Bytes bytes, Long e) { + bytes.writeLong(e); + } + + @Override + public long readSize(Bytes bytes) { + return 4L; + } + + @Override + public Long read(Bytes bytes, long size) { + return bytes.readLong(); + } + + @Override + public Long read(Bytes bytes, long size, Long e) { + return bytes.readLong(); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/threadlocal/Provider.java b/src/main/java/net/openhft/chronicle/map/threadlocal/Provider.java new file mode 100644 index 000000000..e8a832b41 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/threadlocal/Provider.java @@ -0,0 +1,85 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.threadlocal; + +public abstract class Provider { + + public static Provider of(Class tClass) { + if (StatefulCopyable.class.isAssignableFrom(tClass)) { + return StatefulProvider.INSTANCE; + } + return StatelessProvider.INSTANCE; + } + + public abstract T get(ThreadLocalCopies copies, T original); + + public abstract ThreadLocalCopies getCopies(ThreadLocalCopies copies); + + private static final class StatefulProvider> extends Provider { + private static final Provider INSTANCE = new StatefulProvider(); + + @Override + public ThreadLocalCopies getCopies(ThreadLocalCopies copies) { + if (copies != null) + return copies; + return ThreadLocalCopies.get(); + } + + @Override + public T get(ThreadLocalCopies copies, T original) { + if (copies.currentlyAccessed.compareAndSet(false, true)) { + Object id = original.stateIdentity(); + int m = copies.mask; + Object[] tab = copies.table; + int i = System.identityHashCode(id) & m; + T copy; + while (true) { + Object idInTable = tab[i]; + if (idInTable == id) { + copy = (T) tab[i + 1]; + break; + } else if (idInTable == null) { + tab[i] = id; + tab[i + 1] = copy = original.copy(); + copies.postInsert(); + break; + } + i = (i + 2) & m; + } + copies.currentlyAccessed.set(false); + return copy; + } else { + throw new IllegalStateException( + "Concurrent or recursive access to ThreadLocalCopies is not allowed"); + } + } + } + + private static final class StatelessProvider extends Provider { + private static final Provider INSTANCE = new StatelessProvider(); + + @Override + public M get(ThreadLocalCopies copies, M original) { + return original; + } + + @Override + public ThreadLocalCopies getCopies(ThreadLocalCopies copies) { + return copies; + } + } +} diff --git a/src/main/java/net/openhft/chronicle/map/threadlocal/StatefulCopyable.java b/src/main/java/net/openhft/chronicle/map/threadlocal/StatefulCopyable.java new file mode 100644 index 000000000..d246fd2b8 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/threadlocal/StatefulCopyable.java @@ -0,0 +1,23 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.threadlocal; + +public interface StatefulCopyable> { + Object stateIdentity(); + + S copy(); +} diff --git a/src/main/java/net/openhft/chronicle/map/threadlocal/ThreadLocalCopies.java b/src/main/java/net/openhft/chronicle/map/threadlocal/ThreadLocalCopies.java new file mode 100644 index 000000000..e6dcec6d6 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/threadlocal/ThreadLocalCopies.java @@ -0,0 +1,73 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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.threadlocal; + +import java.util.concurrent.atomic.AtomicBoolean; + +public final class ThreadLocalCopies { + + private static ThreadLocal states = new ThreadLocal() { + @Override + protected ThreadLocalCopies initialValue() { + return new ThreadLocalCopies(); + } + }; + + static ThreadLocalCopies get() { + return states.get(); + } + + AtomicBoolean currentlyAccessed = new AtomicBoolean(false); + Object[] table; + int size = 0, sizeLimit, mask; + + private ThreadLocalCopies() { + init(32); // 16 entries + } + + void init(int doubledCapacity) { + table = new Object[doubledCapacity]; + sizeLimit = doubledCapacity / 3; // 0.66 fullness + mask = (doubledCapacity - 1) & ~1; + } + + void rehash() { + Object[] oldTab = this.table; + int oldCapacity = oldTab.length; + if (oldCapacity == (1 << 30)) + throw new IllegalStateException("Hash is full"); + init(oldCapacity << 1); + int m = mask; + Object[] tab = this.table; + for (int oldI = 0; oldI < oldCapacity; oldI += 2) { + Object id = oldTab[oldI]; + if (id != null) { + int i = System.identityHashCode(id) & m; + while (tab[i] != null) { + i = (i + 2) & m; + } + tab[i] = id; + tab[i + 1] = oldTab[oldI + 1]; + } + } + } + + void postInsert() { + if (++size > sizeLimit) + rehash(); + } +} diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java index 3f64a8e88..8b1a153dd 100644 --- a/src/test/java/net/openhft/chronicle/map/Builder.java +++ b/src/test/java/net/openhft/chronicle/map/Builder.java @@ -66,14 +66,14 @@ public class Builder { .create(getPersistenceFile()); } - static MapProvider> newShmIntInt( + static MapProvider> newShmIntInt( int size, final ArrayBlockingQueue input, final ArrayBlockingQueue output, final byte localIdentifier, byte externalIdentifier) throws IOException { Replicator queue = QueueReplicator.of(localIdentifier, externalIdentifier, input, output); - final ReplicatedChronicleMap result = - (ReplicatedChronicleMap) ChronicleMapBuilder + final ReplicatedChronicleMap result = + (ReplicatedChronicleMap) ChronicleMapBuilder .of(Integer.class, Integer.class) .entries(size) .addReplicator(queue) @@ -88,10 +88,10 @@ public class Builder { if (q == null) throw new AssertionError(); final QueueReplicator finalQ = q; - return new MapProvider>() { + return new MapProvider>() { @Override - public ReplicatedChronicleMap getMap() { + public ReplicatedChronicleMap getMap() { return result; } diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index cfbfac040..3277f74db 100644 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -514,21 +514,15 @@ public class ChronicleMapTest { for (int j = 1; j <= 3; j++) { for (int i = 0; i < entries; i++) { CharSequence userCS = getUserCharSequence(i); - if (i * 9876 == 829682760) { - System.out.println(1); - } if (j > 1) { - if (i == 1) { - System.out.println(1); - } assertNotNull(userCS.toString(), map.getUsing(userCS, value)); } else { map.acquireUsing(userCS, value); } if (i >= 1) assertTrue(userCS.toString(), map.containsKey(getUserCharSequence(1))); - assertEquals(j - 1, value.getValue()); + assertEquals(userCS.toString(), j - 1, value.getValue()); value.addAtomicValue(1); @@ -1174,23 +1168,23 @@ public class ChronicleMapTest { .entrySize(10); builder.entryAndValueAlignment(NO_ALIGNMENT); testOversizeEntriesPutRemoveReplace( - (VanillaChronicleMap) + (VanillaChronicleMap) builder.create(getPersistenceFile()) ); builder.entryAndValueAlignment(Alignment.OF_4_BYTES); testOversizeEntriesPutRemoveReplace( - (VanillaChronicleMap) + (VanillaChronicleMap) builder.create(getPersistenceFile()) ); builder.entryAndValueAlignment(OF_8_BYTES); testOversizeEntriesPutRemoveReplace( - (VanillaChronicleMap) + (VanillaChronicleMap) builder.create(getPersistenceFile()) ); } public void testOversizeEntriesPutRemoveReplace( - VanillaChronicleMap map) { + VanillaChronicleMap map) { String key = "k"; String oversizedKey = "oversized key"; String value = "v"; diff --git a/src/test/java/net/openhft/chronicle/map/Issue42.java b/src/test/java/net/openhft/chronicle/map/Issue42.java index e9d03840a..aae50126b 100644 --- a/src/test/java/net/openhft/chronicle/map/Issue42.java +++ b/src/test/java/net/openhft/chronicle/map/Issue42.java @@ -1,3 +1,19 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.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 org.junit.Test; diff --git a/src/test/java/net/openhft/chronicle/map/MultiMapTimeBaseReplicationTest.java b/src/test/java/net/openhft/chronicle/map/MultiMapTimeBaseReplicationTest.java index f62efe0d5..0458e408a 100644 --- a/src/test/java/net/openhft/chronicle/map/MultiMapTimeBaseReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/MultiMapTimeBaseReplicationTest.java @@ -33,10 +33,10 @@ public class MultiMapTimeBaseReplicationTest { private Replica.ModificationIterator segmentModificationIterator2; private Replica.ModificationIterator segmentModificationIterator1; - private Builder.MapProvider> mapP1; - private Builder.MapProvider> mapP2; - private ReplicatedChronicleMap map2; - private ReplicatedChronicleMap map1; + private Builder.MapProvider> mapP1; + private Builder.MapProvider> mapP2; + private ReplicatedChronicleMap map2; + private ReplicatedChronicleMap map1; private ArrayBlockingQueue map1ToMap2; diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java index 8dc3428e6..b538f0764 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java @@ -37,7 +37,7 @@ import static org.junit.Assert.assertEquals; public class TCPSocketReplicationBootStrapTests { - private ReplicatedChronicleMap map1; + private ReplicatedChronicleMap map1; private ChronicleMap map2; @Test @@ -89,8 +89,8 @@ public class TCPSocketReplicationBootStrapTests { map1 = newTcpSocketShmIntString((byte) 1, 8079, new InetSocketAddress("localhost", 8076)); ChronicleMapBuilder map2aBuilder = newTcpSocketShmBuilder(Integer.class, CharSequence.class, (byte) 2, 8076); - final ReplicatedChronicleMap map2a = - (ReplicatedChronicleMap) + final ReplicatedChronicleMap map2a = + (ReplicatedChronicleMap) map2aBuilder.create(getPersistenceFile()); map2a.put(10, "EXAMPLE-10"); // this will be the last time that map1 go an update from map2