diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/internal/DataValueBytesMarshallers.java b/src/main/java/net/openhft/chronicle/hash/serialization/internal/DataValueBytesMarshallers.java new file mode 100644 index 000000000..0e71de34d --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/serialization/internal/DataValueBytesMarshallers.java @@ -0,0 +1,481 @@ +/* + * 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.hash.serialization.internal; + +import net.openhft.chronicle.hash.serialization.BytesReader; +import net.openhft.chronicle.hash.serialization.BytesWriter; +import net.openhft.compiler.CompilerUtils; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.ObjectFactory; +import net.openhft.lang.model.*; +import org.jetbrains.annotations.NotNull; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Constructor; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.Map; +import java.util.SortedSet; +import java.util.concurrent.ConcurrentHashMap; + +import static net.openhft.lang.model.DataValueGenerator.*; + +public final class DataValueBytesMarshallers { + + private static final boolean dumpCode = Boolean.getBoolean("dvg.dumpCode"); + private static final Map readersClassMap = + new ConcurrentHashMap<>(); + private static final Map readersWithCustomFactoriesClassMap = + new ConcurrentHashMap<>(); + private static final Map writerClassMap = + new ConcurrentHashMap<>(); + + public static BytesReader acquireBytesReader(Class tClass) { + Class readerClass = acquireReaderClass(tClass); + try { + Field instanceField = readerClass.getDeclaredField("INSTANCE"); + Object reader = instanceField.get(null); + return (BytesReader) reader; + } catch (NoSuchFieldException | IllegalAccessException e) { + throw new AssertionError(e); + } + } + + public static BytesReader acquireBytesReader(Class tClass, ObjectFactory factory) { + Class readerClass = acquireReaderWithCustomFactory(tClass); + try { + Constructor constructor = readerClass.getConstructor(ObjectFactory.class); + Object reader = constructor.newInstance(factory); + return (BytesReader) reader; + } catch (IllegalAccessException | NoSuchMethodException | InstantiationException | + InvocationTargetException e) { + throw new AssertionError(e); + } + } + + public static BytesWriter acquireBytesWriter(Class tClass) { + Class readerClass = acquireWriterClass(tClass); + try { + Field instanceField = readerClass.getDeclaredField("INSTANCE"); + Object reader = instanceField.get(null); + return (BytesWriter) reader; + } catch (NoSuchFieldException | IllegalAccessException e) { + throw new AssertionError(e); + } + } + + + public static Class acquireReaderClass(Class tClass) { + DataValueClasses.directClassFor(tClass); + Class readerClass = readersClassMap.get(tClass); + if (readerClass != null) + return readerClass; + DataValueModel dvmodel = DataValueModels.acquireModel(tClass); + for (Class clazz : dvmodel.nestedModels()) { + // touch them to make sure they are loaded. + Class clazz2 = acquireReaderClass(clazz); + } + String actual = generateBytesReader(tClass); + if (dumpCode) + LoggerFactory.getLogger(DataValueGenerator.class).info(actual); + ClassLoader classLoader = tClass.getClassLoader(); + String className = bytesReaderName(tClass, false); + try { + readerClass = classLoader.loadClass(className); + } catch (ClassNotFoundException ignored) { + try { + readerClass = CompilerUtils.CACHED_COMPILER + .loadFromJava(classLoader, className, actual); + } catch (ClassNotFoundException e) { + throw new AssertionError(e); + } + } + readersClassMap.put(tClass, readerClass); + return readerClass; + } + + public static Class acquireWriterClass(Class tClass) { + Class writerClass = writerClassMap.get(tClass); + if (writerClass != null) + return writerClass; + DataValueModel dvmodel = DataValueModels.acquireModel(tClass); + for (Class clazz : dvmodel.nestedModels()) { + // touch them to make sure they are loaded. + Class clazz2 = acquireWriterClass(clazz); + } + String actual = generateBytesWriter(tClass); + if (dumpCode) + LoggerFactory.getLogger(DataValueGenerator.class).info(actual); + ClassLoader classLoader = tClass.getClassLoader(); + String className = bytesWriterName(tClass, false); + try { + writerClass = classLoader.loadClass(className); + } catch (ClassNotFoundException ignored) { + try { + writerClass = CompilerUtils.CACHED_COMPILER + .loadFromJava(classLoader, className, actual); + } catch (ClassNotFoundException e) { + throw new AssertionError(e); + } + } + writerClassMap.put(tClass, writerClass); + return writerClass; + } + + public static Class acquireReaderWithCustomFactory(Class tClass) { + Class c = readersWithCustomFactoriesClassMap.get(tClass); + if (c != null) + return c; + acquireReaderClass(tClass); + String actual = generateWithCustomFactoryClass(tClass); + if (dumpCode) + LoggerFactory.getLogger(DataValueGenerator.class).info(actual); + ClassLoader classLoader = tClass.getClassLoader(); + String className = bytesReaderName(tClass, false) + "$WithCustomFactory"; + try { + c = classLoader.loadClass(className); + } catch (ClassNotFoundException ignored) { + try { + c = CompilerUtils.CACHED_COMPILER + .loadFromJava(classLoader, className, actual); + } catch (ClassNotFoundException e) { + throw new AssertionError(e); + } + } + readersWithCustomFactoriesClassMap.put(tClass, c); + return c; + } + + private static String generateWithCustomFactoryClass(Class tClass) { + DataValueModel dvModel = DataValueModels.acquireModel(tClass); + + SortedSet imported = newImported(); + imported.add(ObjectFactory.class); + imported.add(NotNull.class); + imported.add(dvModel.type()); + + String simpleName = tClass.getSimpleName(); + + String bytesReaderName = bytesReaderName(simpleName(tClass)); + String name = bytesReaderName + "$WithCustomFactory"; + String withCustomFactoryClass = + "public final class " + name + " extends " + bytesReaderName + " {\n" + + " private static final long serialVersionUID = 0L;\n" + + "\n" + + " @NotNull\n" + + " private final ObjectFactory<" + simpleName + "> factory;\n" + + "\n" + + " " + name + "(@NotNull ObjectFactory<" + + simpleName + "> factory) {\n" + + " this.factory = factory;\n" + + " }\n" + + "\n" + + " @Override\n" + + " " + simpleName + " getInstance() throws Exception {\n" + + " return factory.create();\n" + + " }\n" + + "}\n"; + + StringBuilder sb = new StringBuilder(); + appendPackage(dvModel, sb); + appendImported(imported, sb); + sb.append(withCustomFactoryClass); + return sb.toString(); + } + + + + private static String bytesReaderName(Class type, boolean simple) { + return (simple ? type.getSimpleName() : type.getName()) + "$$BytesReader"; + } + + private static String bytesReaderName(String className) { + return className + "$$BytesReader"; + } + + public static String bytesWriterName(Class type, boolean simple) { + return (simple ? simpleName(type) : type.getName()) + "$$BytesWriter"; + } + + private static String generateBytesReader(Class tClass) { + DataValueModel dvModel = DataValueModels.acquireModel(tClass); + + SortedSet imported = newImported(); + imported.add(BytesReader.class); + imported.add(Bytes.class); + imported.add(ByteableMarshaller.class); + imported.add(Byteable.class); + imported.add(dvModel.type()); + + StringBuilder read = generateReadBody(dvModel, imported); + + StringBuilder sb = new StringBuilder(); + appendPackage(dvModel, sb); + appendImported(imported, sb); + String simpleName = tClass.getSimpleName(); + String bytesReaderName = bytesReaderName(tClass, false); + String simpleReaderName = bytesReaderName(simpleName(tClass)); + sb.append("\npublic class ").append(simpleReaderName) + .append(" implements BytesReader<").append(simpleName).append("> {\n"); + declareSerialVersionUID(sb); + declareStaticInstance(sb, tClass); + generatePrivateConstructor(sb, simpleReaderName); + generateGetInstance(sb, tClass); + generateDelegatingRead(sb, simpleName); + generateRead(read, sb, simpleName, tClass); + sb.append("}\n"); + return sb.toString(); + } + + private static StringBuilder generateReadBody(DataValueModel dvModel, + SortedSet imported) { + StringBuilder read = new StringBuilder(); + + Map.Entry[] entries = + heapSizeOrderedFields(dvModel); + for (Map.Entry entry : entries) { + String name = entry.getKey(); + FieldModel model = entry.getValue(); + Class type = model.type(); + if (shouldImport(type)) + imported.add(type); + + Method setter = getSetter(model); + Method getter = getGetter(model); + + Method orderedSetter = getOrderedSetter(model); + Method volatileGetter = getVolatileGetter(model); + + Method defaultSetter = setter != null ? setter : orderedSetter; + Method defaultGetter = getter != null ? getter : volatileGetter; + if (!dvModel.isScalar(type)) { + if (model.isArray()) { + read.append(" for (int i = 0; i < ") + .append(model.indexSize().value()).append("; i++) {\n"); + read.append(" toReuse.") + .append(defaultSetter.getName()).append("(i, ") + .append(bytesReaderName(type, false)) + .append(".INSTANCE").append(".read(bytes, ") + .append(computeNonScalarOffset(dvModel, type)).append(", ") + .append(defaultGetter.getName()).append("(i)));"); + read.append(" }\n"); + } else { + read.append(" toReuse.") + .append(defaultSetter.getName()).append("(") + .append(bytesReaderName(type, false)) + .append(".INSTANCE").append(".read(bytes, ") + .append(computeNonScalarOffset(dvModel, type)).append(", ") + .append(defaultGetter.getName()).append("()));"); + } + } else { + if (model.isArray()) { + read.append(" for (int i = 0; i < ") + .append(model.indexSize().value()).append("; i++) {\n"); + read.append(" toReuse.").append(defaultSetter.getName()) + .append("(i, bytes.read").append(bytesType(type)).append("());\n"); + read.append(" }\n"); + } else { + read.append(" toReuse.").append(defaultSetter.getName()) + .append("(bytes.read").append(bytesType(type)).append("());\n"); + } + } + } + return read; + } + + private static String generateBytesWriter(Class tClass) { + DataValueModel dvModel = DataValueModels.acquireModel(tClass); + + SortedSet imported = newImported(); + imported.add(BytesWriter.class); + imported.add(Bytes.class); + imported.add(Byteable.class); + imported.add(dvModel.type()); + + String write = generateWriteBody(dvModel, imported); + + StringBuilder sb = new StringBuilder(); + appendPackage(dvModel, sb); + appendImported(imported, sb); + sb.append("public enum ").append(bytesWriterName(tClass, true)) + .append(" implements BytesWriter<").append(tClass.getSimpleName()).append("> {\n"); + sb.append(" INSTANCE;\n\n"); + generateSize(tClass, dvModel, sb); + generateWrite(tClass, dvModel, write, sb); + generateShouldNotBeNull(tClass, sb); + sb.append("}\n"); + return sb.toString(); + } + + private static void generateShouldNotBeNull(Class tClass, StringBuilder sb) { + sb.append(" private static void shouldNotBeNull() {\n"); + sb.append(" throw new NullPointerException(\"Sub-members of " + + tClass.getSimpleName() + " shouldn't be null for this writer. You should specify" + + " custom writer (e. g. using keyMarshallers()/valueMarshallers() methods of" + + "ChronicleMapBuilder, if you want to support null fields/array elements.\");\n"); + sb.append(" }\n"); + } + + private static void generateWrite(Class tClass, DataValueModel dvModel, String write, + StringBuilder sb) { + appendOverride(sb); + sb.append(" public void write(Bytes bytes, ") + .append(tClass.getSimpleName()).append(" e) {\n"); + int size = computeNonScalarOffset(dvModel, tClass); + if (size > 16) { + sb.append(" if (e instanceof Byteable) {\n"); + sb.append( + " Bytes eBytes = ((Byteable) e).bytes();\n" + + " if (eBytes != null) {\n" + + " bytes.write(eBytes, ((Byteable) e).offset(), " + + size + ");\n" + + " } else {\n" + + " throw new NullPointerException(" + + "\"You are trying to write a byteable object of \" +\n" + + " e.getClass() + \", \" +\n" + + " \"which bytes are not assigned. I. e. most likely " + + "the object is uninitialized.\");\n" + + " }\n"); + sb.append(" return;\n"); + sb.append(" }\n"); + } + sb.append(write); + sb.append(" }\n\n"); + } + + private static void generateSize(Class tClass, DataValueModel dvModel, StringBuilder sb) { + appendOverride(sb); + sb.append(" public long size(").append(tClass.getSimpleName()).append(" e) {\n"); + sb.append(" return ").append(computeNonScalarOffset(dvModel, tClass)).append(";\n"); + sb.append(" }\n\n"); + } + + private static String generateWriteBody(DataValueModel dvModel, SortedSet imported) { + StringBuilder write = new StringBuilder(); + + Map.Entry[] entries = + heapSizeOrderedFields(dvModel); + for (Map.Entry entry : entries) { + String name = entry.getKey(); + FieldModel model = entry.getValue(); + Class type = model.type(); + if (shouldImport(type)) + imported.add(type); + + Method setter = getSetter(model); + Method getter = getGetter(model); + + Method orderedSetter = getOrderedSetter(model); + Method volatileGetter = getVolatileGetter(model); + + Method defaultSetter = setter != null ? setter : orderedSetter; + Method defaultGetter = getter != null ? getter : volatileGetter; + if (!dvModel.isScalar(type)) { + if (model.isArray()) { + write.append(" for (int i = 0; i < ") + .append(model.indexSize().value()).append("; i++) {\n"); + write.append(" ") + .append(type.getName()).append(" ").append(" $ = e.") + .append(defaultGetter.getName()).append("(i);\n"); + write.append(" ") + .append("if ($ == null) shouldNotBeNull();\n"); + write.append(" ") + .append(bytesWriterName(type, false)) + .append(".INSTANCE").append(".write(bytes, $);\n"); + write.append(" }\n"); + } else { + write.append(" {") + .append(type.getName()).append(" ").append(" $ = e.") + .append(defaultGetter.getName()).append("();\n"); + write.append(" ") + .append("if ($ == null) shouldNotBeNull();\n"); + write.append(" ") + .append(bytesWriterName(type, false)) + .append(".INSTANCE").append(".write(bytes, $);}\n"); + } + } else { + if (model.isArray()) { + write.append(" for (int i = 0; i < ") + .append(model.indexSize().value()).append("; i++) {\n"); + write.append(" bytes.write").append(bytesType(type)) + .append("(e.").append(defaultGetter.getName()).append("(i));\n"); + write.append(" }\n"); + } else { + write.append(" bytes.write").append(bytesType(type)) + .append("(e.").append(defaultGetter.getName()).append("());\n"); + } + } + } + return write.toString(); + } + + private static void generateRead(StringBuilder read, StringBuilder sb, String simpleName, + Class cl) { + appendOverride(sb); + sb.append(" public ").append(simpleName).append(" read(Bytes bytes, long size, ") + .append(simpleName).append(" toReuse) {\n"); + sb.append(" try {\n"); + sb.append(" if (toReuse == null)\n"); + sb.append(" toReuse = getInstance();\n"); + sb.append(" if (toReuse instanceof Byteable && ") + .append("((Byteable) toReuse).bytes() == null) {\n"); + sb.append(" " + + "ByteableMarshaller.setBytesAndOffset(((Byteable) toReuse), bytes);\n"); + sb.append(" bytes.skip(size);\n"); + sb.append(" return toReuse;\n"); + sb.append(" }\n"); + sb.append(read); + sb.append(" return toReuse;\n"); + sb.append(" } catch (Exception e) {\n"); + sb.append(" throw new IllegalStateException(e);\n"); + sb.append(" }\n"); + sb.append("}\n\n"); + } + + private static void generateDelegatingRead(StringBuilder sb, String simpleName) { + appendOverride(sb); + sb.append(" public ").append(simpleName).append(" read(Bytes bytes, long size) {\n"); + sb.append(" return read(bytes, size, null);\n"); + sb.append(" }\n\n"); + } + + private static void appendOverride(StringBuilder sb) { + sb.append(" @Override\n"); + } + + private static void generatePrivateConstructor(StringBuilder sb, String bytesReaderName) { + sb.append(" ").append(bytesReaderName).append("() {}\n\n"); + } + + private static void declareStaticInstance(StringBuilder sb, Class cl) { + String bytesReaderName = bytesReaderName(simpleName(cl)); + sb.append(" public static final ").append(bytesReaderName).append(" INSTANCE = new ") + .append(bytesReaderName).append("();\n\n"); + } + + private static StringBuilder declareSerialVersionUID(StringBuilder sb) { + return sb.append(" private static final long serialVersionUID = 0L;\n\n"); + } + + private static void generateGetInstance(StringBuilder sb, Class cl) { + sb.append(" ").append(cl.getCanonicalName()) + .append(" getInstance() throws Exception {\n"); + sb.append(" ").append("return new ").append(cl.getName()).append("$$Native();\n"); + sb.append(" ").append("}\n\n"); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java index d3bffe682..a9eb45115 100755 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -148,8 +148,6 @@ abstract class AbstractChronicleMapBuilder keyClass, Class valueClass) { keyBuilder = new SerializationBuilder(keyClass, SerializationBuilder.Role.KEY); valueBuilder = new SerializationBuilder(valueClass, SerializationBuilder.Role.VALUE); - if (valueClass == LongValue.class) - prepareValueBytesOnAcquire(new ZeroOutValueBytes(valueSize())); } protected static boolean offHeapReference(Class valueClass) { diff --git a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java index 4f34d14f4..34818fe75 100644 --- a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.map; +import net.openhft.lang.model.*; import net.openhft.lang.threadlocal.Provider; import net.openhft.lang.threadlocal.ThreadLocalCopies; import net.openhft.chronicle.hash.serialization.*; @@ -28,7 +29,6 @@ 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.impl.*; -import net.openhft.lang.model.Byteable; import net.openhft.lang.values.LongValue; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -37,7 +37,13 @@ import java.io.Externalizable; import java.io.IOException; import java.io.Serializable; import java.lang.reflect.Modifier; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.net.URL; +import java.util.Arrays; +import java.util.List; +import static net.openhft.chronicle.hash.serialization.SizeMarshallers.constant; import static net.openhft.chronicle.map.Objects.hash; import static net.openhft.chronicle.hash.serialization.SizeMarshallers.stopBit; import static net.openhft.chronicle.map.SerializationBuilder.Role.KEY; @@ -54,8 +60,13 @@ final class SerializationBuilder implements Cloneable, Serializable{ Externalizable.class.isAssignableFrom(c); } + private static final List knownJDKImmutableClasses = Arrays.asList( + String.class, Byte.class, Short.class, Character.class, Integer.class, + Float.class, Long.class, Double.class, BigDecimal.class, BigInteger.class, URL.class + ); + private static boolean instancesAreMutable(Class c) { - return c != String.class; + return !knownJDKImmutableClasses.contains(c); } enum Role {KEY, VALUE} @@ -65,6 +76,7 @@ final class SerializationBuilder implements Cloneable, Serializable{ private final Role role; final Class eClass; private boolean instancesAreMutable; + private boolean dataValueClass = false; private SizeMarshaller sizeMarshaller = stopBit(); private BytesReader reader; private Object interop; @@ -83,6 +95,22 @@ final class SerializationBuilder implements Cloneable, Serializable{ new NewInstanceObjectFactory(eClass) : NullObjectFactory.of(); + if (eClass.isInterface()) { + try { + BytesReader reader = DataValueBytesMarshallers.acquireBytesReader(eClass); + BytesWriter writer = DataValueBytesMarshallers.acquireBytesWriter(eClass); + DataValueModel model = DataValueModels.acquireModel(eClass); + int size = DataValueGenerator.computeNonScalarOffset(model, eClass); + dataValueClass = true; + reader(reader); + writer(writer); + sizeMarshaller(constant((long) size)); + return; + } catch (Exception e) { + // ignore, fall through + } + } + if (concreteClass(eClass) && Byteable.class.isAssignableFrom(eClass)) { ByteableMarshaller byteableMarshaller = ByteableMarshaller.of((Class) eClass); sizeMarshaller(byteableMarshaller); @@ -129,6 +157,10 @@ final class SerializationBuilder implements Cloneable, Serializable{ } } + private void clearDefaults() { + dataValueClass = false; + } + @SuppressWarnings("unchecked") private BytesMarshaller chooseMarshaller(Class eClass, Class classForMarshaller) { if (BytesMarshallable.class.isAssignableFrom(eClass)) @@ -144,6 +176,7 @@ final class SerializationBuilder implements Cloneable, Serializable{ } public SerializationBuilder interop(BytesInterop interop) { + clearDefaults(); return copyingInterop(null) .setInterop(interop) .metaInterop(DelegatingMetaBytesInterop.>instance()) @@ -152,6 +185,7 @@ final class SerializationBuilder implements Cloneable, Serializable{ } public SerializationBuilder writer(BytesWriter writer) { + clearDefaults(); if (writer instanceof BytesInterop) return interop((BytesInterop) writer); return copyingInterop(CopyingInterop.FROM_WRITER) @@ -161,6 +195,7 @@ final class SerializationBuilder implements Cloneable, Serializable{ } public SerializationBuilder marshaller(BytesMarshaller marshaller) { + clearDefaults(); return copyingInterop(CopyingInterop.FROM_MARSHALLER) .reader(BytesReaders.fromBytesMarshaller(marshaller)) .setInterop(marshaller) @@ -247,6 +282,7 @@ final class SerializationBuilder implements Cloneable, Serializable{ } public SerializationBuilder reader(BytesReader reader) { + clearDefaults(); this.reader = reader; return this; } @@ -282,6 +318,10 @@ final class SerializationBuilder implements Cloneable, Serializable{ @SuppressWarnings("unchecked") public SerializationBuilder factory(ObjectFactory factory) { + if (dataValueClass) { + reader(DataValueBytesMarshallers.acquireBytesReader(eClass, factory)); + return this; + } if (!marshallerUseFactory(eClass)) { throw new IllegalStateException("Default marshaller for " + eClass + " value don't use object factory"); diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index b1191f790..a6b6ba4af 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -62,11 +62,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private final String name; private final AbstractChronicleMapBuilder chronicleMapBuilder; - private ByteBuffer outBuffer; - private ByteBufferBytes outBytes; + private volatile ByteBuffer outBuffer; + private volatile ByteBufferBytes outBytes; - private ByteBuffer inBuffer; - private ByteBufferBytes inBytes; + private volatile ByteBuffer inBuffer; + private volatile ByteBufferBytes inBytes; private final ReaderWithSize keyReaderWithSize; private final WriterWithSize keyWriterWithSize; @@ -82,12 +82,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private final boolean putReturnsNull; private final boolean removeReturnsNull; + private final ReentrantLock inBytesLock = new ReentrantLock(true); private final ReentrantLock outBytesLock = new ReentrantLock(true); private ExecutorService executorService; - static enum EventId { HEARTBEAT, STATEFUL_UPDATE, @@ -163,9 +163,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return lazyExecutorService().submit(new Callable() { @Override public V call() throws Exception { - - V oldValue = StatelessChronicleMap.this.put(key, value); - return oldValue; + return StatelessChronicleMap.this.put(key, value); } }); } @@ -175,8 +173,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return lazyExecutorService().submit(new Callable() { @Override public V call() throws Exception { - V oldValue = StatelessChronicleMap.this.remove(key); - return oldValue; + return StatelessChronicleMap.this.remove(key); } }); } @@ -204,8 +201,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return executorService; } - private SocketChannel lazyConnect(final long timeoutMs, - final InetSocketAddress remoteAddress) throws IOException { + private synchronized SocketChannel lazyConnect(final long timeoutMs, + final InetSocketAddress remoteAddress) throws IOException { if (LOG.isDebugEnabled()) LOG.debug("attempting to connect to " + remoteAddress); @@ -245,14 +242,14 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * java.net.InetSocketAddress)} will attempt to establish the connection when the client make * the first map method call. * - * @param remoteAddress + * @param remoteAddress the Inet Socket Address * @return * @throws java.io.IOException * @see net.openhft.chronicle.map.StatelessChronicleMap#lazyConnect(long, * java.net.InetSocketAddress) */ - private void attemptConnect(final InetSocketAddress remoteAddress) throws IOException { + private synchronized void attemptConnect(final InetSocketAddress remoteAddress) throws IOException { // ensures that the excising connection are closed closeExisting(); @@ -282,10 +279,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * -127 ( when the server receives this it knows its dealing with a stateless client, receive * back an identifier from the server * - * @param clientChannel + * @param clientChannel clientChannel * @throws java.io.IOException */ - private void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException { + private synchronized void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException { connectionByte[0] = STATELESS_CLIENT_IDENTIFIER; this.connectionOutBuffer.clear(); @@ -354,7 +351,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - public V putIfAbsent(K key, V value) { + public V putIfAbsent(@NotNull K key, V value) { if (key == null || value == null) throw new NullPointerException(); @@ -363,7 +360,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - public boolean remove(Object key, Object value) { + public boolean remove(@NotNull Object key, Object value) { if (key == null) throw new NullPointerException(); @@ -375,14 +372,14 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - public boolean replace(K key, V oldValue, V newValue) { + public boolean replace(@NotNull K key, @NotNull V oldValue, @NotNull V newValue) { if (key == null || oldValue == null || newValue == null) throw new NullPointerException(); return fetchBoolean(REPLACE_WITH_OLD_AND_NEW_VALUE, key, oldValue, newValue); } - public V replace(K key, V value) { + public V replace(@NotNull K key, @NotNull V value) { if (key == null || value == null) throw new NullPointerException(); @@ -428,13 +425,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchObject(String.class, TO_STRING); } - public synchronized boolean isEmpty() { + public boolean isEmpty() { return fetchBoolean(IS_EMPTY); } public boolean containsKey(Object key) { - return fetchBoolean(CONTAINS_KEY, (K) key); + return fetchBooleanK(CONTAINS_KEY, (K) key); } private NullPointerException keyNotNullNPE() { @@ -442,7 +439,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } public boolean containsValue(Object value) { - return fetchBoolean(CONTAINS_VALUE, (V) value); + return fetchBooleanV(CONTAINS_VALUE, (V) value); } public long longSize() { @@ -503,41 +500,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - public synchronized void putAll(Map map) { - - final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : - writeEventAnSkip(PUT_ALL); - - writeEntriesForPutAll(map); - - if (putReturnsNull) - sendWithoutAcc(outBytes, sizeLocation); - else - blockingFetch(sizeLocation); - } - - private synchronized void sendWithoutAcc(final Bytes bytes, long sizeLocation) { - writeSizeAt(sizeLocation); - long timeoutTime = config.timeoutMs() + System.currentTimeMillis(); - try { - for (; ; ) { - if (clientChannel == null) - clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); - - try { - send(bytes, timeoutTime); - return; - } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { - checkTimeout(timeoutTime); - clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); - } - } - } catch (IOException e) { - close(); - throw new IORuntimeException(e); - } - } - private synchronized void writeEntriesForPutAll(Map map) { final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; @@ -599,9 +561,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon /** * estimates the size based on what been completed so far - * - * @param numberOfEntries - * @param numberOfEntriesReadSoFar */ private long estimateSize(int numberOfEntries, int numberOfEntriesReadSoFar) { // we will back the size estimate on what we have so far @@ -649,7 +608,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private void resizeBufferInBuffer(int size, long start) { if (LOG.isDebugEnabled()) - LOG.debug("resizing buffer to size=" + size); + LOG.debug("InBuffer resizing buffer to size=" + size); if (size < inBuffer.capacity()) throw new IllegalStateException("it not possible to resize the buffer smaller"); @@ -685,9 +644,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon inBytes.position(start); } - public synchronized void clear() { - // get the data back from the server - blockingFetch(writeEventAnSkip(CLEAR)); + public void clear() { + + fetchVoid(CLEAR); } @NotNull @@ -697,7 +656,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long startTime = System.currentTimeMillis(); final long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); - final long transactionId = blockingSend(sizeLocation, startTime); + final long transactionId = send(sizeLocation, startTime); // get the data back from the server final Collection result = new ArrayList(); @@ -734,7 +693,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long sizeLocation = writeEventAnSkip(ENTRY_SET); final long startTime = System.currentTimeMillis(); final long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); - final long transactionId = blockingSend(sizeLocation, startTime); + final long transactionId = send(sizeLocation, startTime); // get the data back from the server ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); @@ -772,13 +731,82 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return result.entrySet(); } + public void putAll(@NotNull Map map) { + + + final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : + writeEventAnSkip(PUT_ALL); + + final long startTime = System.currentTimeMillis(); + final long timeoutTime = startTime + config.timeoutMs(); + final long transactionId; + final int numberOfEntries = map.size(); + int numberOfEntriesReadSoFar = 0; + + outBytesLock.lock(); + try { + + outBytes.writeStopBit(numberOfEntries); + assert outBytes.limit() == outBytes.capacity(); + ThreadLocalCopies copies = keyWriterWithSize.getCopies(null); + final Object keyWriter = keyWriterWithSize.writerForLoop(copies); + copies = valueWriterWithSize.getCopies(copies); + final Object valueWriter = valueWriterWithSize.writerForLoop(copies); + + for (final Map.Entry e : map.entrySet()) { + final K key = e.getKey(); + final V value = e.getValue(); + if (key == null || value == null) + throw new NullPointerException(); + + numberOfEntriesReadSoFar++; + + long start = outBytes.position(); + + // putAll if a bit more complicated than the others + // as the volume of data could cause us to have to resize our buffers + resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar, start); + + final Class keyClass = key.getClass(); + if (!kClass.isAssignableFrom(keyClass)) { + throw new ClassCastException("key=" + key + " is of type=" + keyClass + " " + + "and should" + + " be of type=" + kClass); + } + + writeKeyInLoop(key, keyWriter, copies); + + final Class valueClass = value.getClass(); + if (!vClass.isAssignableFrom(valueClass)) + throw new ClassCastException("value=" + value + " is of type=" + valueClass + + " and " + + "should be of type=" + vClass); + + writeValueInLoop(value, valueWriter, copies); + + final int len = (int) (outBytes.position() - start); + + if (len > maxEntrySize) + maxEntrySize = len; + } + + transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); + } + + + if (!putReturnsNull) + blockingFetchReadOnly(timeoutTime, transactionId); + } + @NotNull public Set keySet() { final long sizeLocation = writeEventAnSkip(KEY_SET); final long startTime = System.currentTimeMillis(); final long timeoutTime = startTime + config.timeoutMs(); - final long transactionId = blockingSend(sizeLocation, startTime); + final long transactionId = send(sizeLocation, startTime); final Set result = new HashSet<>(); final BytesReader keyReader = keyReaderWithSize.readerForLoop(null); @@ -827,16 +855,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytes.clear(); outBytes.write((byte) event.ordinal()); - long sizeLocation = markSizeLocation(); - return sizeLocation; + return markSizeLocation(); } /** * skips for the transactionid - * - * @param event - * @return */ private long writeEventAnSkip(EventId event) { final long sizeLocation = writeEvent(event); @@ -898,22 +922,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - private Bytes blockingFetch(long sizeLocation) { - try { - long startTime = System.currentTimeMillis(); - return blockingFetchThrowable(sizeLocation, - nextUniqueTransaction(startTime), startTime + this.config.timeoutMs()); - } catch (IOException e) { - close(); - throw new IORuntimeException(e); - } catch (InterruptedException e) { - close(); - throw new RuntimeException(e); - } catch (Exception e) { - close(); - throw e; - } - } /** * sends data to the server via TCP/IP @@ -922,7 +930,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @param startTime the current time * @return a unique transaction id */ - private long blockingSend(long sizeLocation, final long startTime) { + private long send(long sizeLocation, final long startTime) { long transactionId = nextUniqueTransaction(startTime); final long timeoutTime = startTime + this.config.timeoutMs(); try { @@ -940,6 +948,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // send out all the bytes send(outBytes, timeoutTime); + outBytes.clear(); + outBytes.buffer().clear(); break; } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { @@ -957,32 +967,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return transactionId; } - - private Bytes blockingFetchThrowable(long sizeLocation, final long transactionId, final long timeoutTime) throws IOException, InterruptedException { - - for (; ; ) { - if (clientChannel == null) - clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); - - try { - - if (LOG.isDebugEnabled()) - LOG.debug("sending data with transactionId=" + transactionId); - - writeSizeAndTransactionIdAt(sizeLocation, transactionId); - - // send out all the bytes - send(outBytes, timeoutTime); - - return blockingFetchThrowable(timeoutTime, transactionId); - } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { - checkTimeout(timeoutTime); - clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); - } - } - } - - private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) { try { return blockingFetchThrowable(timeoutTime, transactionId); @@ -995,88 +979,99 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } catch (Exception e) { close(); throw new RuntimeException(e); + } catch (AssertionError e) { + LOG.error("", e); + throw e; } } - + // this is a transactionid and size that has been read by another thread. private volatile long parkedTransactionId; private volatile int parkedRemainingBytes; private volatile long parkedTransactionTimeStamp; - ByteBufferBytes sizeAndTransactionIdBuffer = new ByteBufferBytes(ByteBuffer.allocate - (SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID)); - - - private synchronized Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, + private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, InterruptedException { int remainingBytes = 0; - for (; ; ) { - // read the next item from the socket - if (parkedTransactionId == 0) { + try { + for (; ; ) { - assert parkedTransactionTimeStamp == 0; - assert parkedRemainingBytes == 0; + // read the next item from the socket + if (parkedTransactionId == 0) { - receive(SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID, timeoutTime); - - final int messageSize = inBytes.readInt(); - final int remainingBytes0 = messageSize - (SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID); - final long transactionId0 = inBytes.readLong(); - assert transactionId0 != 0; + assert parkedTransactionTimeStamp == 0; + assert parkedRemainingBytes == 0; - // if the transaction id is not for this thread, park it and read the next one - if (transactionId0 == transactionId) { + receive(SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID, timeoutTime); - // we have the correct transaction id ! - parkedTransactionId = 0; - remainingBytes = remainingBytes0; - assert remainingBytes > 0; + final int messageSize = inBytes.readInt(); + final int remainingBytes0 = messageSize - (SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID); + final long transactionId0 = inBytes.readLong(); + assert transactionId0 != 0; + + + // check the transaction id is reasonable + assert String.valueOf(transactionId0).startsWith("14"); + + + // if the transaction id is not for this thread, park it and read the next one + if (transactionId0 == transactionId) { + + // we have the correct transaction id ! + parkedTransactionId = 0; + remainingBytes = remainingBytes0; + assert remainingBytes > 0; + + clearParked(); + + break; + + } else { + + parkedTransactionTimeStamp = System.currentTimeMillis(); + parkedRemainingBytes = remainingBytes0; + parkedTransactionId = transactionId0; + + pause(); + continue; + } + } + + // the transaction id was read by another thread, but it is for this thread + if (parkedTransactionId == transactionId) { + remainingBytes = parkedRemainingBytes; + clearParked(); + break; + } + + + // time out the old transaction id + if (System.currentTimeMillis() - timeoutTime > + parkedTransactionTimeStamp) { + + LOG.error("", new IllegalStateException("Skipped Message with transaction-id=" + + parkedTransactionTimeStamp + + ", this can occur when you have another thread which has called the " + + "stateless client and terminated abruptly before the message has been " + + "returned from the server")); + + // read the the next message + receive(parkedRemainingBytes, timeoutTime); clearParked(); - break; - - } else { - - parkedTransactionTimeStamp = System.currentTimeMillis(); - parkedRemainingBytes = remainingBytes0; - parkedTransactionId = transactionId0; - - pause(); - continue; } + + pause(); } - - // the transaction id was read by another thread, but it is for this thread - if (parkedTransactionId == transactionId) { - remainingBytes = parkedRemainingBytes; - clearParked(); - break; - } - - - // time out the old transaction id - if (System.currentTimeMillis() - timeoutTime > - parkedTransactionTimeStamp) { - - LOG.error("", new IllegalStateException("Skipped Message with transaction-id=" + - parkedTransactionTimeStamp + - ", this can occur when you have another thread which has called the " + - "stateless client and terminated abruptly before the message has been " + - "returned from the server")); - parkedTransactionTimeStamp = 0; - parkedTransactionId = 0; - parkedRemainingBytes = 0; - - // read the the next message - receive(remainingBytes, timeoutTime); - continue; - - } - - pause(); + } catch (Exception | Error e) { + parkedTransactionTimeStamp = 0; + parkedTransactionId = 0; + parkedRemainingBytes = 0; + LOG.error("", e); + throw e; } @@ -1127,6 +1122,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private void pause() throws InterruptedException { + + /// inBytesLock.writeLock().isHeldByCurrentThread() don't call this as it not atomic inBytesLock.unlock(); inBytesLock.lock(); } @@ -1142,10 +1139,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon */ private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { - - inBytes.position(0); - inBytes.limit(requiredNumberOfBytes); - inBytes.buffer().position(0); inBytes.buffer().limit(requiredNumberOfBytes); @@ -1189,14 +1182,19 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long size = outBytes.position() - locationOfSize; final long pos = outBytes.position(); outBytes.position(locationOfSize); - outBytes.writeInt((int) size - SIZE_OF_SIZE); + int size0 = (int) size - SIZE_OF_SIZE; + assert size0 > 0; + + outBytes.writeInt(size0); outBytes.writeLong(transactionId); outBytes.position(pos); } private void writeSizeAt(long locationOfSize) { final long size = outBytes.position() - locationOfSize; - outBytes.writeInt(locationOfSize, (int) size - SIZE_OF_SIZE); + int size0 = (int) size - SIZE_OF_SIZE; + assert size0 > 0; + outBytes.writeInt(locationOfSize, size0); } private long markSizeLocation() { @@ -1292,7 +1290,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon int size = Integer.parseInt(substring.substring(0, i)); long requiresExtra = size - outBytes.remaining(); - System.out.println(size + substring); resizeBufferOutBuffer((int) (outBytes.capacity() + requiresExtra), start); } else resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start); @@ -1318,7 +1315,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // receive inBytesLock.lock(); try { - return (O) (blockingFetchReadOnly(timeoutTime, transactionId).readObject()); + return (O) blockingFetchReadOnly(timeoutTime, transactionId).readObject(); } finally { inBytesLock.unlock(); } @@ -1350,25 +1347,21 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private long send(final EventId eventId, final long startTime) { - long transactionId = nextUniqueTransaction(startTime); - // send outBytesLock.lock(); try { final long sizeLocation = writeEventAnSkip(eventId); - transactionId = blockingSend(sizeLocation, startTime); + return send(sizeLocation, startTime); } finally { outBytesLock.unlock(); } - - return transactionId; } private V readValue(ThreadLocalCopies copies, Bytes in) { return valueReaderWithSize.readNullable(in, copies); } - private boolean fetchBoolean(final EventId eventId, K key, V... values) { + private boolean fetchBoolean(final EventId eventId, K key, V value) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1376,15 +1369,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = writeEventAnSkip(eventId); - copies = writeKey(key); - for (V value : values) { - copies = writeValue(value, copies); - } - transactionId = blockingSend(sizeLocation, startTime); + writeValue(value, copies); + + transactionId = send(sizeLocation, startTime); } finally { outBytesLock.unlock(); } @@ -1394,18 +1384,21 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - - private boolean fetchBoolean(final EventId eventId, V value) { + private boolean fetchBoolean(final EventId eventId, K key, V value1, V value2) { final long startTime = System.currentTimeMillis(); + ThreadLocalCopies copies; long transactionId; outBytesLock.lock(); try { - final long sizeLocation = writeEventAnSkip(eventId); - writeValue(value, null); - transactionId = blockingSend(sizeLocation, startTime); + + + copies = writeKey(key); + copies = writeValue(value1, copies); + writeValue(value2, copies); + transactionId = send(sizeLocation, startTime); } finally { outBytesLock.unlock(); } @@ -1415,6 +1408,41 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } + private boolean fetchBooleanV(final EventId eventId, V value) { + final long startTime = System.currentTimeMillis(); + + long transactionId; + + outBytesLock.lock(); + try { + final long sizeLocation = writeEventAnSkip(eventId); + writeValue(value, null); + transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); + } + + return readBoolean(transactionId, startTime); + } + + + private boolean fetchBooleanK(final EventId eventId, K key) { + final long startTime = System.currentTimeMillis(); + + long transactionId; + + outBytesLock.lock(); + try { + final long sizeLocation = writeEventAnSkip(eventId); + writeKey(key, null); + transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); + } + + return readBoolean(transactionId, startTime); + } + private long fetchLong(final EventId eventId) { final long startTime = System.currentTimeMillis(); @@ -1423,16 +1451,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = writeEventAnSkip(eventId); - - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } finally { outBytesLock.unlock(); } return readLong(transactionId, startTime); - } private boolean fetchBoolean(final EventId eventId) { @@ -1442,17 +1467,42 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = writeEventAnSkip(eventId); - - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } finally { outBytesLock.unlock(); } - return readBoolean(transactionId, startTime); + } + private void fetchVoid(final EventId eventId) { + + final long startTime = System.currentTimeMillis(); + + long transactionId; + + outBytesLock.lock(); + try { + final long sizeLocation = writeEventAnSkip(eventId); + transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); + } + + readVoid(transactionId, startTime); + } + + private void readVoid(long transactionId, long startTime) { + long timeoutTime = startTime + this.config.timeoutMs(); + + // receive + inBytesLock.lock(); + try { + blockingFetchReadOnly(timeoutTime, transactionId); + } finally { + inBytesLock.unlock(); + } } @@ -1480,14 +1530,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { final long sizeLocation = writeEventAnSkip(eventId); - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } finally { outBytesLock.unlock(); } - return readInt(transactionId, startTime); - } @@ -1511,6 +1559,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.unlock(); } + if (eventReturnsNull(eventId)) return null; @@ -1521,6 +1570,38 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon "supported"); } + private R fetchObject(Class rClass, final EventId eventId, K key, V value1, V value2) { + final long startTime = System.currentTimeMillis(); + ThreadLocalCopies copies; + + long transactionId; + + outBytesLock.lock(); + try { + + final long sizeLocation = eventReturnsNull(eventId) + ? writeEvent(eventId) + : writeEventAnSkip(eventId); + + copies = writeKey(key); + copies = writeValue(value1, copies); + copies = writeValue(value2, copies); + + transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); + } + + + if (eventReturnsNull(eventId)) + return null; + + if (rClass == vClass) + return (R) readValue(transactionId, startTime, copies); + else + throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + + "supported"); + } private R fetchObject(Class rClass, final EventId eventId, K key) { final long startTime = System.currentTimeMillis(); @@ -1536,7 +1617,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon : writeEventAnSkip(eventId); copies = writeKey(key); - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } finally { outBytesLock.unlock(); } @@ -1551,38 +1632,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon "supported"); } - private O fetchObject(final EventId eventId) { - final long startTime = System.currentTimeMillis(); - long transactionId = send(eventId, startTime); - return readObject(transactionId, startTime); - } - - private O fetchObject(final EventId eventId, K key, V... values) { - final long startTime = System.currentTimeMillis(); - long transactionId = sendAndGetTransactionId(startTime, eventId, key, values); - return (O) readObject(transactionId, startTime); - } - - private long sendAndGetTransactionId(long startTime, final EventId eventId, K key, V... values) { - ThreadLocalCopies copies; - long transactionId;// send - outBytesLock.lock(); - try { - final long sizeLocation = (eventReturnsNull(eventId)) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); - - copies = writeKey(key); - for (V value : values) { - copies = writeValue(value, copies); - } - transactionId = blockingSend(sizeLocation, startTime); - } finally { - outBytesLock.unlock(); - } - return transactionId; - } - private boolean eventReturnsNull(EventId eventId) { @@ -1625,7 +1674,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long sizeLocation = writeEventAnSkip(eventId); writeKey(key); writeObject(object); - transactionId = blockingSend(sizeLocation, startTime); + transactionId = send(sizeLocation, startTime); } finally { outBytesLock.unlock(); } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index bb8e4d6e5..e380aca09 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1182,6 +1182,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable state = out.readByte(); sizeInBytes = out.readInt(); + assert sizeInBytes >= 0; + // if the buffer is too small to read this payload we will have to grow the // size of the buffer long requiredSize = sizeInBytes + SIZE_OF_SIZE + 1; @@ -1202,6 +1204,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } final long nextEntryPos = out.position() + sizeInBytes; + assert nextEntryPos > 0; final long limit = out.limit(); out.limit(nextEntryPos); @@ -1524,6 +1527,8 @@ class StatelessServerConnector { writeException(writer, e); writeSizeAndFlags(sizeLocation + SIZE_OF_TRANSACTION_ID, true, writer); + + e.printStackTrace(); return null; } @@ -1868,6 +1873,8 @@ class StatelessServerConnector { long pos = out.position(); long limit = out.limit(); + // System.out.println("Sending with size=" + size); + if (LOG.isDebugEnabled()) { out.position(locationOfSize); out.limit(pos); diff --git a/src/test/java/eg/TestIntrumentVOInterface.java b/src/test/java/eg/TestIntrumentVOInterface.java new file mode 100644 index 000000000..915080bac --- /dev/null +++ b/src/test/java/eg/TestIntrumentVOInterface.java @@ -0,0 +1,45 @@ +package eg; + +import net.openhft.lang.model.constraints.MaxSize; + +/** + * Created by Vanitha on 12/5/2014. + */ +public interface TestIntrumentVOInterface { + + + public int getSizeOfInstrumentIDArray(); + + public void setSizeOfInstrumentIDArray(int sizeOfInstrumentIDArray); + + + public String getSymbol(); + + public void setSymbol(@MaxSize(20) String symbol); + + + + + public String getCurrencyCode(); + + public void setCurrencyCode(@MaxSize(4) String currencyCode); + + + public void setInstrumentIDAt(@MaxSize(2) int location, TestInstrumentIDVOInterface instrumentID); + + public TestInstrumentIDVOInterface getInstrumentIDAt(int location); + + + interface TestInstrumentIDVOInterface { + + + public String getIdSource(); + + public void setIdSource(@MaxSize(6) String idSource); + + public String getSecurityId(); + + public void setSecurityId(@MaxSize(100) String securityId); + + } +} diff --git a/src/test/java/eg/TestNesterInterfaceMarshal.java b/src/test/java/eg/TestNesterInterfaceMarshal.java new file mode 100644 index 000000000..cf36ad7cc --- /dev/null +++ b/src/test/java/eg/TestNesterInterfaceMarshal.java @@ -0,0 +1,110 @@ +package eg; + +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.map.ChronicleMap; +import net.openhft.chronicle.map.ChronicleMapBuilder; +import net.openhft.lang.model.DataValueClasses; +import org.junit.Assert; +import org.junit.Ignore; +import org.junit.Test; + +import java.net.InetSocketAddress; +import java.util.concurrent.TimeUnit; + +/** + * Created by Vanitha on 12/5/2014. + */ +public class TestNesterInterfaceMarshal { + + + + + /** + * Test that both maps are created first and then populated the first map so that second will get the contents + * @throws Exception + */ + @Test + public void testReplicatedMap() throws Exception { + ChronicleMap map =createReplicatedMap1(); + ChronicleMap replicatedMap=createReplicatedMap2(); + + + //Store some data into MAP1 + TestIntrumentVOInterface intrumentVOInterface= DataValueClasses.newDirectInstance(TestIntrumentVOInterface.class); + intrumentVOInterface.setSymbol("FIXFlyer"); + intrumentVOInterface.setCurrencyCode("USA"); + intrumentVOInterface.setSizeOfInstrumentIDArray(2); + + TestIntrumentVOInterface.TestInstrumentIDVOInterface instrumentIDVOInterface= intrumentVOInterface.getInstrumentIDAt(0); + instrumentIDVOInterface.setIdSource("CUSIP"); + instrumentIDVOInterface.setSecurityId("TEST"); + intrumentVOInterface.setInstrumentIDAt(0,instrumentIDVOInterface); + map.put("KEY1",intrumentVOInterface); + + + + + Assert.assertNotNull(map); + Assert.assertNotNull(replicatedMap); + + + int t = 0; + for (; t < 5000; t++) { + if (map.equals(replicatedMap)) + break; + Thread.sleep(1); + } + + Assert.assertEquals(map, replicatedMap); + Assert.assertTrue(!replicatedMap.isEmpty()); + + System.out.println(map.get("KEY1")); + System.out.println(replicatedMap.get("KEY1")); + + map.close(); + replicatedMap.close(); + + } + + protected ChronicleMap createReplicatedMap1() { + + ChronicleMap chronicleMap = null; + TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig=TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS); + try { + ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestIntrumentVOInterface.class) + .entries(5000L).entrySize(2000) + .replication((byte) 1, tcpTransportAndNetworkConfig); + + + chronicleMap = builder.create(); + + } catch (Exception e) { + System.out.println("Error(s) creating instrument cache: " + e); + } + return chronicleMap; + } + + protected ChronicleMap createReplicatedMap2() { + + ChronicleMap chronicleMap = null; + TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig=TcpTransportAndNetworkConfig.of(8077,new InetSocketAddress("127.0.0.1",8076)).heartBeatInterval(1L, TimeUnit.SECONDS); + + try { + ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestIntrumentVOInterface.class) + .putReturnsNull(true) + .removeReturnsNull(true) + .entries(5000L).entrySize(2000) + .replication((byte)2,tcpTransportAndNetworkConfig) ; + + + chronicleMap = builder.create(); + + } catch (Exception e) { + System.out.println("*********************Error(s) creating launcher instrument cache: " + e); + } + return chronicleMap; + } + +} + + diff --git a/src/test/java/net/openhft/chronicle/map/DataKeyValueTest.java b/src/test/java/net/openhft/chronicle/map/DataKeyValueTest.java new file mode 100644 index 000000000..7d2cb72e1 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/DataKeyValueTest.java @@ -0,0 +1,56 @@ +/* + * 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.lang.model.Byteable; +import net.openhft.lang.model.DataValueClasses; +import net.openhft.lang.values.IntValue; +import net.openhft.lang.values.LongValue; +import org.junit.Assert; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class DataKeyValueTest { + + @Test + public void dataKeyValueTest() { + ChronicleMap map = + OnHeapUpdatableChronicleMapBuilder.of(IntValue.class, LongValue.class) + .entries(1000).create(); + IntValue heapKey = DataValueClasses.newInstance(IntValue.class); + IntValue directKey = DataValueClasses.newDirectInstance(IntValue.class); + LongValue heapValue = DataValueClasses.newInstance(LongValue.class); + LongValue directValue = DataValueClasses.newDirectInstance(LongValue.class); + + heapKey.setValue(1); + heapValue.setValue(1); + map.put(heapKey, heapValue); + assertEquals(1, map.get(heapKey).getValue()); + assertEquals(1, map.getUsing(heapKey, directValue).getValue()); + + ((Byteable) directValue).bytes(null, 0L); + directKey.setValue(1); + map.getUsing(directKey, directValue).addValue(1); + assertEquals(2, map.getUsing(directKey, heapValue).getValue()); + + directKey.setValue(2); + heapValue.setValue(3); + map.put(directKey, heapValue); + assertEquals(3, map.get(directKey).getValue()); + } +} diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index b105949ca..e67144d16 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -26,7 +26,8 @@ import org.junit.Test; import java.io.IOException; import java.net.InetSocketAddress; import java.util.*; -import java.util.concurrent.ExecutionException; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -507,6 +508,21 @@ public class StatelessClientTest { } } + @Test(timeout = 10000) + public void testGetAndEntryWeDontHave() throws IOException, + InterruptedException, ExecutionException { + + try (ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) { + try (ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { + assertEquals(null, statelessMap.get(3)); + + } + } + } + @Test(timeout = 10000) public void testEquals() throws IOException, InterruptedException { @@ -547,4 +563,99 @@ public class StatelessClientTest { serverMap1.close(); serverMap2.close(); } -} \ No newline at end of file + + + @Test + public void testThreadSafeness() throws IOException, InterruptedException { + + int nThreads = 1; + final ExecutorService executorService = Executors.newFixedThreadPool(nThreads); + + int count = 1000; + final CountDownLatch latch = new CountDownLatch(count * 2); + final AtomicInteger got = new AtomicInteger(); + + long startTime = System.currentTimeMillis(); + // server + try (ChronicleMap server = ChronicleMapBuilder.of(Integer.class, Integer.class) + .putReturnsNull(true) + .replication((byte) 1, TcpTransportAndNetworkConfig.of(8059)).create()) { + + // stateless client + try (ChronicleMap client = ChronicleMapBuilder.of(Integer.class, + Integer.class) + .statelessClient(new InetSocketAddress("localhost", 8059)).create()) { + + for (int i = 0; i < count; i++) { + + + final int j = i; + executorService.submit(new Runnable() { + @Override + public void run() { + try { + client.put(j, j); + latch.countDown(); + } catch (Error | Exception e) { + e.printStackTrace(); + System.exit(0); + } + } + }); + } + + for (int i = 0; i < count; i++) { + final int j = i; + + executorService.submit(new Runnable() { + @Override + public void run() { + try { + + Integer result = client.get(j); + + + if (result == null) { + executorService.submit(this); + return; + } + + if (result.equals(j)) { + got.incrementAndGet(); + } else { + System.out.println("expected j=" + j + " but got back=" + result); + } + + latch.countDown(); + } catch (Error | Exception e) { + e.printStackTrace(); + System.exit(0); + } + } + }); + + + } + + latch.await(); + System.out.println("" + count + " messages took " + + TimeUnit.MILLISECONDS.toSeconds(System + .currentTimeMillis() - startTime) + " seconds, using " + nThreads + "" + + " threads"); + + assertEquals(count, got.get()); + + } + } finally { + executorService.shutdownNow(); + executorService.awaitTermination(100, TimeUnit.SECONDS); + + } + + + } + + +} + + diff --git a/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$$Heap.java b/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$$Heap.java new file mode 100644 index 000000000..3db5fc71a --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$$Heap.java @@ -0,0 +1,174 @@ +/* + * 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.fromdocs; + +import static net.openhft.lang.Compare.*; +import java.io.IOException; +import net.openhft.chronicle.map.fromdocs.BondVOInterface; +import net.openhft.chronicle.map.fromdocs.BondVOInterface.MarketPx; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.BytesMarshallable; +import net.openhft.lang.model.Copyable; + +public class BondVOInterface$$Heap implements BondVOInterface, BytesMarshallable, Copyable { + private double _coupon; + private long _issueDate; + private long _maturityDate; + private int _entry; + private net.openhft.chronicle.map.fromdocs.BondVOInterface.MarketPx[] _marketPxIntraDayHistory = new net.openhft.chronicle.map.fromdocs.BondVOInterface.MarketPx[7]; + { + for (int i = 0; i < _marketPxIntraDayHistory.length; i++) + _marketPxIntraDayHistory[i] = new net.openhft.chronicle.map.fromdocs.BondVOInterface$MarketPx$$Heap(); + } private java.lang.String _symbol; + + public void setCoupon(double $) { + _coupon = $; + } + + public double getCoupon() { + return _coupon; + } + + public synchronized double addAtomicCoupon(double $) { + return _coupon += $; + } + + public synchronized boolean compareAndSwapCoupon(double _1, double _2) { + if (_coupon == _1) { + _coupon = _2; + return true; + } + return false; + } + public void setIssueDate(long $) { + _issueDate = $; + } + + public long getIssueDate() { + return _issueDate; + } + + public void setMaturityDate(long $) { + _maturityDate = $; + } + + public long getMaturityDate() { + return _maturityDate; + } + + public synchronized long addAtomicMaturityDate(long $) { + return _maturityDate += $; + } + + public void unlockEntry() { + throw new UnsupportedOperationException(); + } public void busyLockEntry() { + throw new UnsupportedOperationException(); + } public void setMarketPxIntraDayHistoryAt(int i, net.openhft.chronicle.map.fromdocs.BondVOInterface.MarketPx $) { + if(i<0) throw new ArrayIndexOutOfBoundsException(i + " must be greater than 0"); + if(i>=7) throw new ArrayIndexOutOfBoundsException(i + " must be less than 7"); + _marketPxIntraDayHistory[i] = $; + } + + public net.openhft.chronicle.map.fromdocs.BondVOInterface.MarketPx getMarketPxIntraDayHistoryAt(int i) { + if(i<0) throw new ArrayIndexOutOfBoundsException(i + " must be greater than 0"); + if(i>=7) throw new ArrayIndexOutOfBoundsException(i + " must be less than 7"); + return _marketPxIntraDayHistory[i]; + } + + public void setSymbol(java.lang.String $) { + _symbol = $; + } + + public java.lang.String getSymbol() { + return _symbol; + } + + @SuppressWarnings("unchecked") + public void copyFrom(net.openhft.chronicle.map.fromdocs.BondVOInterface from) { + setCoupon(from.getCoupon()); + setIssueDate(from.getIssueDate()); + setMaturityDate(from.getMaturityDate()); + for (int i = 0; i < 7; i++){ + setMarketPxIntraDayHistoryAt(i, from.getMarketPxIntraDayHistoryAt(i)); + } + setSymbol(from.getSymbol()); + } + + public void writeMarshallable(Bytes out) { + out.writeDouble(getCoupon()); + out.writeLong(getIssueDate()); + out.writeLong(getMaturityDate()); + for (int i = 0; i < 7; i++){ + out.writeObject(getMarketPxIntraDayHistoryAt(i)); + } + out.writeUTFΔ(getSymbol()); + } + public void readMarshallable(Bytes in) { + _coupon = in.readDouble(); + _issueDate = in.readLong(); + _maturityDate = in.readLong(); + _entry = in.readInt(); + for (int i = 0; i < 7; i++){ + _marketPxIntraDayHistory[i] = in.readObject(net.openhft.chronicle.map.fromdocs.BondVOInterface.MarketPx.class); + } + _symbol = in.readUTFΔ(); + } + + public long longHashCode_marketPxIntraDayHistory() { + long hc = 0; + for (int i = 0; i < 7; i++) { + hc += calcLongHashCode(getMarketPxIntraDayHistoryAt(i)); + } + return hc; + } + + public int hashCode() { + long lhc = longHashCode(); + return (int) ((lhc >>> 32) ^ lhc); + } + + public long longHashCode() { + return ((((calcLongHashCode(getCoupon())) * 10191 + + calcLongHashCode(getIssueDate())) * 10191 + + calcLongHashCode(getMaturityDate())) * 10191 + + longHashCode_marketPxIntraDayHistory()) * 10191 + + calcLongHashCode(getSymbol()); + } + + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof BondVOInterface)) return false; + BondVOInterface that = (BondVOInterface) o; + + if(!isEqual(getCoupon(), that.getCoupon())) return false; + if(!isEqual(getIssueDate(), that.getIssueDate())) return false; + if(!isEqual(getMaturityDate(), that.getMaturityDate())) return false; + for (int i = 0; i <7; i++) { + if(!isEqual(getMarketPxIntraDayHistoryAt(i), that.getMarketPxIntraDayHistoryAt(i))) return false; + } + if(!isEqual(getSymbol(), that.getSymbol())) return false; + return true; + } + + public String toString() { + StringBuilder sb = new StringBuilder(); + sb.append("BondVOInterface{ "); + sb.append(" }"); + return sb.toString(); + } +} diff --git a/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$$Native.java b/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$$Native.java new file mode 100644 index 000000000..6ab936a20 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$$Native.java @@ -0,0 +1,203 @@ +/* + * 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.fromdocs; + + +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.BytesMarshallable; +import net.openhft.lang.model.Byteable; +import net.openhft.lang.model.Copyable; + +import static net.openhft.lang.Compare.calcLongHashCode; +import static net.openhft.lang.Compare.isEqual; + +public class BondVOInterface$$Native + implements BondVOInterface, BytesMarshallable, Byteable, + Copyable { + private static final int COUPON = 0; + private static final int ISSUEDATE = 8; + private static final int MATURITYDATE = 16; + private static final int ENTRY = 24; + private static final int MARKETPXINTRADAYHISTORY = 28; + private final BondVOInterface$MarketPx$$Native _marketPxIntraDayHistory[] = + new BondVOInterface$MarketPx$$Native[7]; + { + for (int i = 0; i < 7; i++) + _marketPxIntraDayHistory[i] = new BondVOInterface$MarketPx$$Native(); + } + private static final int SYMBOL = 308; + + + private Bytes _bytes; + private long _offset; + + + + public void setCoupon(double $) { + _bytes.writeDouble(_offset + COUPON, $); + } + + public double getCoupon() { + return _bytes.readDouble(_offset + COUPON); + } + + public double addAtomicCoupon(double $) { + return _bytes.addAtomicDouble(_offset + COUPON, $); + } public boolean compareAndSwapCoupon(double _1, double _2) { + return _bytes.compareAndSwapDouble(_offset + COUPON, _1, _2); + } + + public void setIssueDate(long $) { + _bytes.writeLong(_offset + ISSUEDATE, $); + } + + public long getIssueDate() { + return _bytes.readLong(_offset + ISSUEDATE); + } + + + + public void setMaturityDate(long $) { + _bytes.writeLong(_offset + MATURITYDATE, $); + } + + public long getMaturityDate() { + return _bytes.readLong(_offset + MATURITYDATE); + } + + public long addAtomicMaturityDate(long $) { + return _bytes.addAtomicLong(_offset + MATURITYDATE, $); + } public void unlockEntry() { + _bytes.unlockInt(_offset + ENTRY); + } public void busyLockEntry() throws InterruptedException { + _bytes.busyLockInt(_offset + ENTRY); + } public void setMarketPxIntraDayHistoryAt(int i, BondVOInterface.MarketPx $) { + _marketPxIntraDayHistory[i].copyFrom($); + } + + public BondVOInterface.MarketPx getMarketPxIntraDayHistoryAt(int i) { + return _marketPxIntraDayHistory[i]; + } + + + + public void setSymbol(java.lang.String $) { + _bytes.writeUTFΔ(_offset + SYMBOL, 20, $); + } + + public java.lang.String getSymbol() { + return _bytes.readUTFΔ(_offset + SYMBOL); + } + + @Override + public void copyFrom(BondVOInterface from) { + setCoupon(from.getCoupon()); + setIssueDate(from.getIssueDate()); + setMaturityDate(from.getMaturityDate()); + for (int i = 0; i < 7; i++){ + setMarketPxIntraDayHistoryAt(i, from.getMarketPxIntraDayHistoryAt(i)); + } + setSymbol(from.getSymbol()); + } + + @Override + public void writeMarshallable(Bytes out) { + out.writeDouble(getCoupon()); + out.writeLong(getIssueDate()); + out.writeLong(getMaturityDate()); + for (int i = 0; i < 7; i++){ + _marketPxIntraDayHistory[i].writeMarshallable(out); + } + out.writeUTFΔ(getSymbol()); + } + @Override + public void readMarshallable(Bytes in) { + setCoupon(in.readDouble()); + setIssueDate(in.readLong()); + setMaturityDate(in.readLong()); + for (int i = 0; i < 7; i++){ + _marketPxIntraDayHistory[i].readMarshallable(in); + } + setSymbol(in.readUTFΔ()); + } + @Override + public void bytes(Bytes bytes, long offset) { + this._bytes = bytes; + this._offset = offset; + for (int i = 0; i < 7; i++){ + ((Byteable) _marketPxIntraDayHistory[i]).bytes(bytes, + _offset + MARKETPXINTRADAYHISTORY + (i * 40)); + } + } + @Override + public Bytes bytes() { + return _bytes; + } + @Override + public long offset() { + return _offset; + } + @Override + public int maxSize() { + return 328; + } + + public long longHashCode_marketPxIntraDayHistory() { + long hc = 0; + for (int i = 0; i < 7; i++) { + hc += calcLongHashCode(getMarketPxIntraDayHistoryAt(i)); + } + return hc; + } + + public int hashCode() { + long lhc = longHashCode(); + return (int) ((lhc >>> 32) ^ lhc); + } + + public long longHashCode() { + return ((((calcLongHashCode(getCoupon())) * 10191 + + calcLongHashCode(getIssueDate())) * 10191 + + calcLongHashCode(getMaturityDate())) * 10191 + + longHashCode_marketPxIntraDayHistory()) * 10191 + + calcLongHashCode(getSymbol()); + } + + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof BondVOInterface)) return false; + BondVOInterface that = (BondVOInterface) o; + + if(!isEqual(getCoupon(), that.getCoupon())) return false; + if(!isEqual(getIssueDate(), that.getIssueDate())) return false; + if(!isEqual(getMaturityDate(), that.getMaturityDate())) return false; + for (int i = 0; i <7; i++) { + if(!isEqual(getMarketPxIntraDayHistoryAt(i), that.getMarketPxIntraDayHistoryAt(i))) + return false; + } + if(!isEqual(getSymbol(), that.getSymbol())) return false; + return true; + } + + public String toString() { + if (_bytes == null) return "bytes is null"; + StringBuilder sb = new StringBuilder(); + sb.append("BondVOInterface{ "); + sb.append(" }"); + return sb.toString(); + } +} diff --git a/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$MarketPx$$Heap.java b/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$MarketPx$$Heap.java new file mode 100644 index 000000000..cd7839302 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$MarketPx$$Heap.java @@ -0,0 +1,128 @@ +/* + * 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.fromdocs; + +import static net.openhft.lang.Compare.*; +import java.io.IOException; +import net.openhft.chronicle.map.fromdocs.BondVOInterface.MarketPx; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.BytesMarshallable; +import net.openhft.lang.model.Copyable; + +public class BondVOInterface$MarketPx$$Heap implements MarketPx, BytesMarshallable, Copyable { + private double _askPx; + private double _bidPx; + private double _callPx; + private double _maturityPx; + private double _parPx; + + public void setAskPx(double $) { + _askPx = $; + } + + public double getAskPx() { + return _askPx; + } + + public void setBidPx(double $) { + _bidPx = $; + } + + public double getBidPx() { + return _bidPx; + } + + public void setCallPx(double $) { + _callPx = $; + } + + public double getCallPx() { + return _callPx; + } + + public void setMaturityPx(double $) { + _maturityPx = $; + } + + public double getMaturityPx() { + return _maturityPx; + } + + public void setParPx(double $) { + _parPx = $; + } + + public double getParPx() { + return _parPx; + } + + @SuppressWarnings("unchecked") + public void copyFrom(net.openhft.chronicle.map.fromdocs.BondVOInterface.MarketPx from) { + setAskPx(from.getAskPx()); + setBidPx(from.getBidPx()); + setCallPx(from.getCallPx()); + setMaturityPx(from.getMaturityPx()); + setParPx(from.getParPx()); + } + + public void writeMarshallable(Bytes out) { + out.writeDouble(getAskPx()); + out.writeDouble(getBidPx()); + out.writeDouble(getCallPx()); + out.writeDouble(getMaturityPx()); + out.writeDouble(getParPx()); + } + public void readMarshallable(Bytes in) { + _askPx = in.readDouble(); + _bidPx = in.readDouble(); + _callPx = in.readDouble(); + _maturityPx = in.readDouble(); + _parPx = in.readDouble(); + } + public int hashCode() { + long lhc = longHashCode(); + return (int) ((lhc >>> 32) ^ lhc); + } + + public long longHashCode() { + return ((((calcLongHashCode(getAskPx())) * 10191 + + calcLongHashCode(getBidPx())) * 10191 + + calcLongHashCode(getCallPx())) * 10191 + + calcLongHashCode(getMaturityPx())) * 10191 + + calcLongHashCode(getParPx()); + } + + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof BondVOInterface.MarketPx)) return false; + BondVOInterface.MarketPx that = (BondVOInterface.MarketPx) o; + + if(!isEqual(getAskPx(), that.getAskPx())) return false; + if(!isEqual(getBidPx(), that.getBidPx())) return false; + if(!isEqual(getCallPx(), that.getCallPx())) return false; + if(!isEqual(getMaturityPx(), that.getMaturityPx())) return false; + if(!isEqual(getParPx(), that.getParPx())) return false; + return true; + } + + public String toString() { + StringBuilder sb = new StringBuilder(); + sb.append("BondVOInterface.MarketPx{ "); + sb.append(" }"); + return sb.toString(); + } +} \ No newline at end of file diff --git a/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$MarketPx$$Native.java b/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$MarketPx$$Native.java new file mode 100644 index 000000000..67aa52ba0 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/fromdocs/BondVOInterface$MarketPx$$Native.java @@ -0,0 +1,165 @@ +/* + * 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.fromdocs; + +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.BytesMarshallable; +import net.openhft.lang.model.Byteable; +import net.openhft.lang.model.Copyable; + +import static net.openhft.lang.Compare.calcLongHashCode; +import static net.openhft.lang.Compare.isEqual; + +public class BondVOInterface$MarketPx$$Native + implements BondVOInterface.MarketPx, BytesMarshallable, Byteable, + Copyable { + private static final int ASKPX = 0; + private static final int BIDPX = 8; + private static final int CALLPX = 16; + private static final int MATURITYPX = 24; + private static final int PARPX = 32; + + + private Bytes _bytes; + private long _offset; + + + + public void setAskPx(double $) { + _bytes.writeDouble(_offset + ASKPX, $); + } + + public double getAskPx() { + return _bytes.readDouble(_offset + ASKPX); + } + + + + public void setBidPx(double $) { + _bytes.writeDouble(_offset + BIDPX, $); + } + + public double getBidPx() { + return _bytes.readDouble(_offset + BIDPX); + } + + + + public void setCallPx(double $) { + _bytes.writeDouble(_offset + CALLPX, $); + } + + public double getCallPx() { + return _bytes.readDouble(_offset + CALLPX); + } + + + + public void setMaturityPx(double $) { + _bytes.writeDouble(_offset + MATURITYPX, $); + } + + public double getMaturityPx() { + return _bytes.readDouble(_offset + MATURITYPX); + } + + + + public void setParPx(double $) { + _bytes.writeDouble(_offset + PARPX, $); + } + + public double getParPx() { + return _bytes.readDouble(_offset + PARPX); + } + + @Override + public void copyFrom(BondVOInterface.MarketPx from) { + setAskPx(from.getAskPx()); + setBidPx(from.getBidPx()); + setCallPx(from.getCallPx()); + setMaturityPx(from.getMaturityPx()); + setParPx(from.getParPx()); + } + + @Override + public void writeMarshallable(Bytes out) { + out.writeDouble(getAskPx()); + out.writeDouble(getBidPx()); + out.writeDouble(getCallPx()); + out.writeDouble(getMaturityPx()); + out.writeDouble(getParPx()); + } + @Override + public void readMarshallable(Bytes in) { + setAskPx(in.readDouble()); + setBidPx(in.readDouble()); + setCallPx(in.readDouble()); + setMaturityPx(in.readDouble()); + setParPx(in.readDouble()); + } + @Override + public void bytes(Bytes bytes, long offset) { + this._bytes = bytes; + this._offset = offset; + } + @Override + public Bytes bytes() { + return _bytes; + } + @Override + public long offset() { + return _offset; + } + @Override + public int maxSize() { + return 40; + } + public int hashCode() { + long lhc = longHashCode(); + return (int) ((lhc >>> 32) ^ lhc); + } + + public long longHashCode() { + return ((((calcLongHashCode(getAskPx())) * 10191 + + calcLongHashCode(getBidPx())) * 10191 + + calcLongHashCode(getCallPx())) * 10191 + + calcLongHashCode(getMaturityPx())) * 10191 + + calcLongHashCode(getParPx()); + } + + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof BondVOInterface.MarketPx)) return false; + BondVOInterface.MarketPx that = (BondVOInterface.MarketPx) o; + + if(!isEqual(getAskPx(), that.getAskPx())) return false; + if(!isEqual(getBidPx(), that.getBidPx())) return false; + if(!isEqual(getCallPx(), that.getCallPx())) return false; + if(!isEqual(getMaturityPx(), that.getMaturityPx())) return false; + if(!isEqual(getParPx(), that.getParPx())) return false; + return true; + } + + public String toString() { + if (_bytes == null) return "bytes is null"; + StringBuilder sb = new StringBuilder(); + sb.append("BondVOInterface.MarketPx{ "); + sb.append(" }"); + return sb.toString(); + } +}