Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Merge remote-tracking branch 'origin/master'

Browse Source 
Conflicts:
	src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
	src/test/java/eg/BGChronicleTest.java
This commit is contained in:
Peter Lawrey
2014-12-06 00:06:51 +02:00
parent 345496c84d d8f381a8dc
commit e2fcf91e28
13 changed files with 1821 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/serialization/internal/DataValueBytesMarshallers.java
+481
View File
@@ -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<Class, Class> readersClassMap =
new ConcurrentHashMap<>();
private static final Map<Class, Class> readersWithCustomFactoriesClassMap =
new ConcurrentHashMap<>();
private static final Map<Class, Class> writerClassMap =
new ConcurrentHashMap<>();
public static <T> BytesReader<T> acquireBytesReader(Class<T> tClass) {
Class readerClass = acquireReaderClass(tClass);
try {
Field instanceField = readerClass.getDeclaredField("INSTANCE");
Object reader = instanceField.get(null);
return (BytesReader<T>) reader;
} catch (NoSuchFieldException | IllegalAccessException e) {
throw new AssertionError(e);
}
}
public static <T> BytesReader<T> acquireBytesReader(Class<T> tClass, ObjectFactory<T> factory) {
Class readerClass = acquireReaderWithCustomFactory(tClass);
try {
Constructor constructor = readerClass.getConstructor(ObjectFactory.class);
Object reader = constructor.newInstance(factory);
return (BytesReader<T>) reader;
} catch (IllegalAccessException | NoSuchMethodException | InstantiationException |
InvocationTargetException e) {
throw new AssertionError(e);
}
}
public static <T> BytesWriter<T> acquireBytesWriter(Class<T> tClass) {
Class readerClass = acquireWriterClass(tClass);
try {
Field instanceField = readerClass.getDeclaredField("INSTANCE");
Object reader = instanceField.get(null);
return (BytesWriter<T>) reader;
} catch (NoSuchFieldException | IllegalAccessException e) {
throw new AssertionError(e);
}
}
public static <T> Class acquireReaderClass(Class<T> tClass) {
DataValueClasses.directClassFor(tClass);
Class readerClass = readersClassMap.get(tClass);
if (readerClass != null)
return readerClass;
DataValueModel<T> 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 <T> Class acquireWriterClass(Class<T> tClass) {
Class writerClass = writerClassMap.get(tClass);
if (writerClass != null)
return writerClass;
DataValueModel<T> 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 <T> Class acquireReaderWithCustomFactory(Class<T> 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<Class> 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<Class> 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<Class> imported) {
StringBuilder read = new StringBuilder();
Map.Entry<String, FieldModel>[] entries =
heapSizeOrderedFields(dvModel);
for (Map.Entry<String, FieldModel> 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<Class> 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<Class> imported) {
StringBuilder write = new StringBuilder();
Map.Entry<String, FieldModel>[] entries =
heapSizeOrderedFields(dvModel);
for (Map.Entry<String, FieldModel> 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");
}
}
src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java
-2
View File
@@ -148,8 +148,6 @@ abstract class AbstractChronicleMapBuilder<K, V,
AbstractChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
keyBuilder = new SerializationBuilder<K>(keyClass, SerializationBuilder.Role.KEY);
valueBuilder = new SerializationBuilder<V>(valueClass, SerializationBuilder.Role.VALUE);
if (valueClass == LongValue.class)
prepareValueBytesOnAcquire(new ZeroOutValueBytes<K, V>(valueSize()));
}
protected static boolean offHeapReference(Class valueClass) {
src/main/java/net/openhft/chronicle/map/SerializationBuilder.java
+42 -2
View File
@@ -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<E> implements Cloneable, Serializable{
Externalizable.class.isAssignableFrom(c);
}
private static final List<Class> knownJDKImmutableClasses = Arrays.<Class>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<E> implements Cloneable, Serializable{
private final Role role;
final Class<E> eClass;
private boolean instancesAreMutable;
private boolean dataValueClass = false;
private SizeMarshaller sizeMarshaller = stopBit();
private BytesReader<E> reader;
private Object interop;
@@ -83,6 +95,22 @@ final class SerializationBuilder<E> implements Cloneable, Serializable{
new NewInstanceObjectFactory<E>(eClass) :
NullObjectFactory.<E>of();
if (eClass.isInterface()) {
try {
BytesReader<E> reader = DataValueBytesMarshallers.acquireBytesReader(eClass);
BytesWriter<E> writer = DataValueBytesMarshallers.acquireBytesWriter(eClass);
DataValueModel<E> 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<E> implements Cloneable, Serializable{
}
}
private void clearDefaults() {
dataValueClass = false;
}
@SuppressWarnings("unchecked")
private BytesMarshaller<E> chooseMarshaller(Class<E> eClass, Class<E> classForMarshaller) {
if (BytesMarshallable.class.isAssignableFrom(eClass))
@@ -144,6 +176,7 @@ final class SerializationBuilder<E> implements Cloneable, Serializable{
}
public SerializationBuilder<E> interop(BytesInterop<E> interop) {
clearDefaults();
return copyingInterop(null)
.setInterop(interop)
.metaInterop(DelegatingMetaBytesInterop.<E, BytesInterop<E>>instance())
@@ -152,6 +185,7 @@ final class SerializationBuilder<E> implements Cloneable, Serializable{
}
public SerializationBuilder<E> writer(BytesWriter<E> writer) {
clearDefaults();
if (writer instanceof BytesInterop)
return interop((BytesInterop<E>) writer);
return copyingInterop(CopyingInterop.FROM_WRITER)
@@ -161,6 +195,7 @@ final class SerializationBuilder<E> implements Cloneable, Serializable{
}
public SerializationBuilder<E> marshaller(BytesMarshaller<E> marshaller) {
clearDefaults();
return copyingInterop(CopyingInterop.FROM_MARSHALLER)
.reader(BytesReaders.fromBytesMarshaller(marshaller))
.setInterop(marshaller)
@@ -247,6 +282,7 @@ final class SerializationBuilder<E> implements Cloneable, Serializable{
}
public SerializationBuilder<E> reader(BytesReader<E> reader) {
clearDefaults();
this.reader = reader;
return this;
}
@@ -282,6 +318,10 @@ final class SerializationBuilder<E> implements Cloneable, Serializable{
@SuppressWarnings("unchecked")
public SerializationBuilder<E> factory(ObjectFactory<E> 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");
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+297 -248
View File
@@ -62,11 +62,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K> keyReaderWithSize;
private final WriterWithSize<K> keyWriterWithSize;
@@ -82,12 +82,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, Closeable, Clon
return lazyExecutorService().submit(new Callable<V>() {
@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<K, V> implements ChronicleMap<K, V>, Closeable, Clon
return lazyExecutorService().submit(new Callable<V>() {
@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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
public synchronized void putAll(Map<? extends K, ? extends V> 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<? extends K, ? extends V> map) {
final int numberOfEntries = map.size();
int numberOfEntriesReadSoFar = 0;
@@ -599,9 +561,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<V> result = new ArrayList<V>();
@@ -734,7 +693,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, Closeable, Clon
return result.entrySet();
}
public void putAll(@NotNull Map<? extends K, ? extends V> 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<? extends K, ? extends V> 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<K> 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<K> result = new HashSet<>();
final BytesReader<K> keyReader = keyReaderWithSize.readerForLoop(null);
@@ -827,16 +855,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, Closeable, Clon
outBytesLock.unlock();
}
if (eventReturnsNull(eventId))
return null;
@@ -1521,6 +1570,38 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
"supported");
}
private <R> R fetchObject(Class<R> 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> R fetchObject(Class<R> rClass, final EventId eventId, K key) {
final long startTime = System.currentTimeMillis();
@@ -1536,7 +1617,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
: writeEventAnSkip(eventId);
copies = writeKey(key);
transactionId = blockingSend(sizeLocation, startTime);
transactionId = send(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
@@ -1551,38 +1632,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
"supported");
}
private <O> O fetchObject(final EventId eventId) {
final long startTime = System.currentTimeMillis();
long transactionId = send(eventId, startTime);
return readObject(transactionId, startTime);
}
private <O> 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<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final long sizeLocation = writeEventAnSkip(eventId);
writeKey(key);
writeObject(object);
transactionId = blockingSend(sizeLocation, startTime);
transactionId = send(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+7
View File
@@ -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<K, V> {
writeException(writer, e);
writeSizeAndFlags(sizeLocation + SIZE_OF_TRANSACTION_ID, true, writer);
e.printStackTrace();
return null;
}
@@ -1868,6 +1873,8 @@ class StatelessServerConnector<K, V> {
long pos = out.position();
long limit = out.limit();
// System.out.println("Sending with size=" + size);
if (LOG.isDebugEnabled()) {
out.position(locationOfSize);
out.limit(pos);