Merge remote-tracking branch 'origin/master'

This commit is contained in:
Rob Austin
2014-10-03 08:19:44 +01:00
12 changed files with 373 additions and 68 deletions
@@ -18,6 +18,9 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.map.serialization.MetaBytesInterop;
import net.openhft.chronicle.map.serialization.MetaBytesWriter;
import net.openhft.chronicle.map.serialization.MetaProvider;
import net.openhft.chronicle.map.threadlocal.Provider;
import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies;
import net.openhft.lang.Maths;
import net.openhft.lang.io.*;
import net.openhft.lang.io.serialization.*;
@@ -66,6 +69,8 @@ public class ChronicleMapBuilder<K, V> implements Cloneable {
private ObjectSerializer objectSerializer;
private MapEventListener<K, V, ChronicleMap<K, V>> eventListener =
MapEventListeners.nop();
private V defaultValue = null;
private DefaultValueProvider<K, V> defaultValueProvider = NullValueProvider.INSTANCE;
ChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
keyBuilder = new SerializationBuilder<K>(keyClass, SerializationBuilder.Role.KEY);
@@ -459,6 +464,65 @@ public class ChronicleMapBuilder<K, V> implements Cloneable {
return eventListener;
}
/**
* Specifies the value to be put for each key queried in {@link ChronicleMap#get get()} and
* {@link ChronicleMap#getUsing(Object, Object) getUsing()} methods, if the key is absent
* in the map. Then this default value is returned from query method.
*
* <p>Setting default value to {@code null} is interpreted as map shouldn't put any default
* value for absent keys. This is by default.
*
* <p>This configuration overrides any previous
* {@link #defaultValueProvider(DefaultValueProvider)} configuration to this
* {@code ChronicleMapBuilder}.
*
* @param defaultValue the default value to be put to the map for absent keys during
* {@code get()} and {@code getUsing()} calls and returned from these calls
* @return this builder object back
*/
public ChronicleMapBuilder<K, V> defaultValue(V defaultValue) {
this.defaultValue = defaultValue;
this.defaultValueProvider = null;
return this;
}
/**
* Specifies the function to obtain a value for the key during {@link ChronicleMap#get get()}
* and {@link ChronicleMap#getUsing(Object, Object) getUsing()} calls, if the key is absent
* in the map. If the obtained value is non-null, it is put for the key in the map and then
* returned from current {@code get()} or {@code getUsing()} call.
*
* <p>This configuration overrides any previous {@link #defaultValue(Object)} configuration
* to this {@code ChronicleMapBuilder}.
*
* @param defaultValueProvider the strategy to obtain a default value by the absent key
* @return this builder object back
*/
public ChronicleMapBuilder<K, V> defaultValueProvider(
DefaultValueProvider<K, V> defaultValueProvider) {
this.defaultValueProvider = defaultValueProvider;
return this;
}
/**
* Non-public because should be called only after {@link #preMapConstruction()}
*/
DefaultValueProvider<K, V> defaultValueProvider() {
if (defaultValueProvider != null)
return defaultValueProvider;
if (defaultValue == null)
return NullValueProvider.INSTANCE;
Object originalValueWriter = valueBuilder.interop();
Provider writerProvider = Provider.of(originalValueWriter.getClass());
ThreadLocalCopies copies = writerProvider.getCopies(null);
Object valueWriter = writerProvider.get(copies, originalValueWriter);
MetaProvider metaWriterProvider = valueBuilder.metaInteropProvider();
copies = metaWriterProvider.getCopies(copies);
MetaBytesWriter metaValueWriter = metaWriterProvider.get(copies,
valueBuilder.metaInterop(), valueWriter, defaultValue);
return new ConstantValueProvider<K, V>(defaultValue, metaValueWriter, valueWriter);
}
public ChronicleMap<K, V> create(File file) throws IOException {
for (int i = 0; i < 10; i++) {
if (file.exists() && file.length() > 0) {
@@ -524,6 +588,9 @@ public class ChronicleMapBuilder<K, V> implements Cloneable {
}
private void preMapConstruction() {
keyBuilder.objectSerializer(objectSerializer());
valueBuilder.objectSerializer(objectSerializer());
int maxSize = entrySize() * figureBufferAllocationFactor();
keyBuilder.maxSize(maxSize);
valueBuilder.maxSize(maxSize);
@@ -0,0 +1,52 @@
/*
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.map;
import net.openhft.chronicle.map.serialization.BytesReader;
import net.openhft.chronicle.map.serialization.MetaBytesWriter;
import net.openhft.lang.io.ByteBufferBytes;
import java.nio.ByteBuffer;
final class ConstantValueProvider<K, V> implements DefaultValueProvider<K, V> {
private static final long serialVersionUID = 0L;
private transient V value;
private final int size;
private final byte[] serializedValueBytes;
<W> ConstantValueProvider(V value, MetaBytesWriter<V, W> metaValueWriter, W valueWriter) {
this.value = value;
size = (int) metaValueWriter.size(valueWriter, value);
serializedValueBytes = new byte[size];
ByteBufferBytes bytes = new ByteBufferBytes(ByteBuffer.wrap(serializedValueBytes));
metaValueWriter.write(valueWriter, bytes, value);
}
boolean wasDeserialized() {
return value == null;
}
void initTransients(BytesReader<V> reader) {
value = reader.read(new ByteBufferBytes(ByteBuffer.wrap(serializedValueBytes)), size);
}
@Override
public V get(K key, V usingValue) {
return value;
}
}
@@ -0,0 +1,45 @@
/*
* 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 java.io.Serializable;
/**
* Default value computation strategy, used
* in {@link ChronicleMapBuilder#defaultValueProvider(DefaultValueProvider)} configuration.
*
* @param <K> map key class
* @param <V> map value class
*/
public interface DefaultValueProvider<K, V> extends Serializable {
/**
* Returns a value to be put and returned from {@link ChronicleMap#get(Object)} get()} or
* {@link ChronicleMap#getUsing(Object, Object)} call for the specified key, if it is absent
* in the map.
*
* <p>If this method returns {@code null}, it isn't put for the key in the map.
*
* @param key key absent in the map
* @param usingValue an object provided as second parameter of current
* {@link ChronicleMap#getUsing(Object, Object)} call, or {@code null}, if this method is called
* within {@code get()} call. This object might be optionally initialized and returned back
* from this method.
* @return value to be put for the specified key in the map, or {@code null}, if the key should
* remain absent in the map
*/
V get(K key, V usingValue);
}
@@ -31,27 +31,15 @@ public abstract class MapEventListener<K, V, M extends ChronicleMap<K, V>>
private static final long serialVersionUID = 0L;
/**
* This is called when there was no existing entry for a key.
* Optionally you can provide a value to add to the map.
* This method is called if the key is found in the map during {@link ChronicleMap#get get},
* {@link ChronicleMap#getUsing getUsing} or {@link ChronicleMap#acquireUsing acquireUsing}
* method call.
*
* @param map accessed
* @param keyBytes bytes of the key looked up
* @param key object used as key
* @param usingValue value provided to reuse, could be null.
* @return null if null should be returned, or a value to put in the map and return.
*/
public V onGetMissing(M map, K key, V usingValue) {
return null;
}
/**
* This method is called if a value is found in the map.
*
* @param map accessed
* @param map the accessed map
* @param entry bytes of the entry
* @param metaDataBytes length of meta data for this map.
* @param key looked up
* @param value found
* @param metaDataBytes length of meta data for this map
* @param key the key looked up
* @param value the value found for the key
*/
public void onGetFound(M map, Bytes entry, int metaDataBytes, K key, V value) {
// do nothing
@@ -35,13 +35,6 @@ public final class MapEventListeners {
private static final long serialVersionUID = 0L;
public final Logger LOGGER = LoggerFactory.getLogger(getClass());
@Override
public Object onGetMissing(ChronicleMap map,
Object key, Object usingValue) {
LOGGER.info("{} missed {}", map.file(), key);
return null;
}
@Override
public void onGetFound(ChronicleMap map, Bytes entry, int metaDataBytes,
Object key, Object value) {
@@ -84,13 +77,6 @@ public final class MapEventListeners {
private static final long serialVersionUID = 0L;
public final Logger LOGGER = LoggerFactory.getLogger(getClass());
@Override
public Object onGetMissing(ChronicleMap map,
Object key, Object usingValue) {
LOGGER.info("{} missed {}", map.file(), key);
return null;
}
@Override
public void onGetFound(ChronicleMap map, Bytes entry, int metaDataBytes,
Object key, Object value) {
@@ -0,0 +1,26 @@
/*
* 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;
enum NullValueProvider implements DefaultValueProvider {
INSTANCE;
@Override
public Object get(Object key, Object usingValue) {
return null;
}
}
@@ -1299,12 +1299,6 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
@Override
public V onGetMissing(ChronicleMap<K, V> map, K key,
V usingValue) {
return nextListener.onGetMissing(map, key, usingValue);
}
@Override
public void onGetFound(ChronicleMap<K, V> map, Bytes entry, int metaDataBytes,
K key, V value) {
@@ -22,17 +22,22 @@ import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.serialization.*;
import net.openhft.lang.io.serialization.impl.*;
import net.openhft.lang.model.Byteable;
import net.openhft.lang.model.DataValueClasses;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import java.io.Externalizable;
import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.Modifier;
import static net.openhft.chronicle.map.Objects.hash;
import static net.openhft.chronicle.map.serialization.SizeMarshallers.stopBit;
final class SerializationBuilder<E> implements Cloneable {
private static boolean concreteClass(Class c) {
return !c.isInterface() && !Modifier.isAbstract(c.getModifiers());
}
private static boolean marshallerUseFactory(Class c) {
return Byteable.class.isAssignableFrom(c) ||
@@ -51,25 +56,22 @@ final class SerializationBuilder<E> implements Cloneable {
private CopyingInterop copyingInterop = null;
private MetaBytesInterop<E, ?> metaInterop;
private MetaProvider<E, ?, ?> metaInteropProvider;
private ObjectFactory<E> factory;
private ObjectFactory<E> factory;
@SuppressWarnings("unchecked")
SerializationBuilder(Class<E> eClass, Role role) {
this.bufferIdentity = role;
this.eClass = eClass;
Class<E> classForMarshaller = marshallerUseFactory(eClass) && eClass.isInterface() ?
DataValueClasses.directClassFor(eClass) : eClass;
ObjectFactory<E> factory = null;
if (role == Role.VALUE) {
factory = marshallerUseFactory(eClass) ?
new AllocateInstanceObjectFactory(
eClass.isInterface() ? classForMarshaller : eClass) :
ObjectFactory<E> factory = this.factory;
if (factory == null && role == Role.VALUE) {
factory = concreteClass(eClass) && marshallerUseFactory(eClass) ?
new AllocateInstanceObjectFactory(eClass) :
NullObjectFactory.INSTANCE;
}
if (Byteable.class.isAssignableFrom(eClass)) {
agileMarshaller(ByteableMarshaller.of((Class) classForMarshaller), factory);
if (concreteClass(eClass) && Byteable.class.isAssignableFrom(eClass)) {
agileMarshaller(ByteableMarshaller.of((Class) eClass), factory);
} else if (eClass == CharSequence.class || eClass == String.class) {
reader((BytesReader<E>) CharSequenceReader.of(), factory);
writer((BytesWriter<E>) CharSequenceWriter.instance());
@@ -78,8 +80,10 @@ final class SerializationBuilder<E> implements Cloneable {
} else if (eClass == byte[].class) {
reader((BytesReader<E>) ByteArrayMarshaller.INSTANCE, factory);
interop((BytesInterop<E>) ByteArrayMarshaller.INSTANCE);
} else {
marshaller(chooseMarshaller(eClass, classForMarshaller), factory);
} else if (concreteClass(eClass)) {
BytesMarshaller<E> marshaller = chooseMarshaller(eClass, eClass);
if (marshaller != null)
marshaller(marshaller, factory);
}
}
@@ -93,7 +97,7 @@ final class SerializationBuilder<E> implements Cloneable {
return (BytesMarshaller<E>) IntegerMarshaller.INSTANCE;
if (eClass == Double.class)
return (BytesMarshaller<E>) DoubleMarshaller.INSTANCE;
return SerializableMarshaller.INSTANCE;
return null;
}
private SerializationBuilder<E> copyingInterop(CopyingInterop copyingInterop) {
@@ -148,6 +152,16 @@ final class SerializationBuilder<E> implements Cloneable {
return this;
}
public SerializationBuilder<E> objectSerializer(ObjectSerializer serializer) {
if (reader == null || interop == null) {
ObjectFactory<E> factory = this.factory;
if (factory == null)
factory = NullObjectFactory.INSTANCE;
marshaller(new SerializableMarshaller(serializer), factory);
}
return this;
}
public SizeMarshaller sizeMarshaller() {
return sizeMarshaller;
}
@@ -297,24 +311,46 @@ final class SerializationBuilder<E> implements Cloneable {
}
private enum SerializableMarshaller implements BytesMarshaller {
INSTANCE;
private static class SerializableMarshaller implements BytesMarshaller {
private static final long serialVersionUID = 0L;
private final ObjectSerializer serializer;
private SerializableMarshaller(ObjectSerializer serializer) {
this.serializer = serializer;
}
@Override
public void write(Bytes bytes, Object obj) {
bytes.writeObject(obj);
try {
serializer.writeSerializable(bytes, obj, null);
} catch (IOException e) {
throw new IllegalStateException(e);
}
}
@Nullable
@Override
public Object read(Bytes bytes) {
return bytes.readObject();
try {
return serializer.readSerializable(bytes, null, null);
} catch (IOException e) {
throw new IllegalStateException(e);
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e);
}
}
@Nullable
@Override
public Object read(Bytes bytes, @Nullable Object obj) {
return bytes.readInstance(null, obj);
try {
return serializer.readSerializable(bytes, null, obj);
} catch (IOException e) {
throw new IllegalStateException(e);
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e);
}
}
}
@@ -72,6 +72,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
final MVW originalMetaValueWriter;
final MetaProvider<V, VW, MVW> metaValueWriterProvider;
final ObjectFactory<V> valueFactory;
final DefaultValueProvider<K, V> defaultValueProvider;
final int metaDataBytes;
// private final int replicas;
@@ -118,6 +119,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
originalMetaValueWriter = (MVW) valueBuilder.metaInterop();
metaValueWriterProvider = (MetaProvider) valueBuilder.metaInteropProvider();
valueFactory = valueBuilder.factory();
defaultValueProvider = builder.defaultValueProvider();
lockTimeOutNS = builder.lockTimeOut(TimeUnit.NANOSECONDS);
@@ -163,6 +165,16 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
valueReaderProvider = Provider.of((Class) originalValueReader.getClass());
valueWriterProvider = Provider.of((Class) originalValueWriter.getClass());
if (defaultValueProvider instanceof ConstantValueProvider) {
ConstantValueProvider<K, V> constantValueProvider =
(ConstantValueProvider<K, V>) defaultValueProvider;
if (constantValueProvider.wasDeserialized()) {
ThreadLocalCopies copies = valueReaderProvider.getCopies(null);
BytesReader<V> valueReader = valueReaderProvider.get(copies, originalValueReader);
constantValueProvider.initTransients(valueReader);
}
}
@SuppressWarnings("unchecked")
Segment[] ss = (Segment[]) Array.newInstance(segmentType(), actualSegments);
this.segments = ss;
@@ -765,7 +777,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
} else {
if (usingValue instanceof Byteable)
((Byteable) usingValue).bytes(null, 0);
return notifyMissed(key, usingValue);
return defaultValueProvider.get(key, usingValue);
}
}
@@ -1070,13 +1082,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
V notifyMissed(K key, V usingValue) {
if (eventListener() != MapEventListeners.NOP) {
return eventListener().onGetMissing(VanillaChronicleMap.this, key, usingValue);
}
return null;
}
void notifyRemoved(long offset, K key, V value, final int pos) {
if (eventListener() != MapEventListeners.NOP) {
tmpBytes.storePositionAndSize(bytes, offset, entrySize);