Merge remote-tracking branch 'origin/master'

This commit is contained in:
Peter Lawrey
2014-10-27 19:49:50 +00:00
35 changed files with 1949 additions and 1153 deletions
@@ -21,7 +21,10 @@ package net.openhft.chronicle.common;
import net.openhft.chronicle.map.*;
import net.openhft.chronicle.set.ChronicleSet;
import net.openhft.chronicle.set.ChronicleSetBuilder;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.serialization.*;
import net.openhft.lang.io.serialization.impl.AllocateInstanceObjectFactory;
import net.openhft.lang.io.serialization.impl.NewInstanceObjectFactory;
import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory;
import net.openhft.lang.model.Byteable;
import org.jetbrains.annotations.NotNull;
@@ -345,6 +348,29 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
*/
B keyMarshaller(@NotNull BytesMarshaller<K> keyMarshaller);
/**
* Configures factory which is used to create a new key instance, if key class is either
* {@link Byteable}, {@link BytesMarshallable} or {@link Externalizable} subclass in maps,
* created by this builder. If {@linkplain #keyMarshaller(BytesMarshaller) custom key
* marshaller} is configured, this configuration is unused, because it is incapsulated in
* {@link BytesMarshaller#read(Bytes)} method (without provided instance to read the data into),
* i. e. it's is the user-side responsibility.
*
* <p>Default key deserialization factory is {@link NewInstanceObjectFactory}, which creates
* a new key instance using {@link Class#newInstance()} default constructor. You could provide
* an {@link AllocateInstanceObjectFactory}, which uses {@code Unsafe.allocateInstance(Class)}
* (you might want to do this for better performance or if you don't want to initialize fields),
* or a factory which calls a key class constructor with some arguments, or a factory which
* internally delegates to instance pool or {@link ThreadLocal}, to reduce allocations.
*
* @param keyDeserializationFactory the key factory used to produce instances to deserialize
* data in
* @return this builder back
* @throws IllegalStateException if custom key marshaller is specified or key class is not
* either {@code Byteable}, {@code BytesMarshallable} or {@code Externalizable}
*/
B keyDeserializationFactory(@NotNull ObjectFactory<K> keyDeserializationFactory);
/**
* Specifies that key objects, queried with the hash containers, created by this builder, are
* inherently immutable. Keys in {@link ChronicleMap} or {@link ChronicleSet} are not required
@@ -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.
*/
/**
* Contains common interfaces and utilities for {@link net.openhft.chronicle.map.ChronicleMap
* ChronicleMaps} (<code>net.openhft.chronicle.map</code> package) and
* {@link net.openhft.chronicle.set.ChronicleSet ChronicleSets}
* (<code>net.openhft.chronicle.set</code> package). This package should be the root of
* {@code net.openhft.chronicle.map} and {@code net.openhft.chronicle.set} packages, moved to
* the same level because {@code net.openhft.chronicle} package is occupied by <a
* href="https://github.com/OpenHFT/Chronicle-Queue">Chronicle Queue</a> project.
*/
package net.openhft.chronicle.common;
@@ -0,0 +1,936 @@
/*
* 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.common.*;
import net.openhft.chronicle.common.serialization.MetaBytesInterop;
import net.openhft.chronicle.common.serialization.MetaBytesWriter;
import net.openhft.chronicle.common.serialization.MetaProvider;
import net.openhft.chronicle.common.threadlocal.Provider;
import net.openhft.chronicle.common.threadlocal.ThreadLocalCopies;
import net.openhft.chronicle.set.ChronicleSetBuilder;
import net.openhft.lang.Maths;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.BytesStore;
import net.openhft.lang.io.DirectStore;
import net.openhft.lang.io.serialization.*;
import net.openhft.lang.io.serialization.impl.AllocateInstanceObjectFactory;
import net.openhft.lang.io.serialization.impl.NewInstanceObjectFactory;
import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory;
import net.openhft.lang.model.Byteable;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import static net.openhft.chronicle.map.Objects.builderEquals;
/**
* {@code AbstractChronicleMapBuilder} manages most of {@link ChronicleMap} configurations; has two
* concrete subclasses: {@link ChronicleMapBuilder} should be used to create maps with ordinary
* values, {@link OffHeapUpdatableChronicleMapBuilder} -- maps with {@link Byteable} values, which
* point directly to off-heap memory; could be used as a classic builder and/or factory. This means
* that in addition to the standard builder usage pattern: <pre>{@code
* ChronicleMap<Key, Value> map = ChronicleMapBuilder
* .of(Key.class, Value.class)
* .entries(100500)
* // ... other configurations
* .create();}</pre>
* one of concrete {@code AbstractChronicleMapBuilder} subclasses, {@link ChronicleMapBuilder} or
* {@link OffHeapUpdatableChronicleMapBuilder}, could be prepared and used to create many similar
* maps: <pre>{@code
* OffHeapUpdatableChronicleMapBuilder<Key, Value> builder = OffHeapUpdatableChronicleMapBuilder
* .of(Key.class, Value.class)
* .entries(100500);
*
* ChronicleMap<Key, Value> map1 = builder.create();
* ChronicleMap<Key, Value> map2 = builder.create();}</pre>
* i. e. created {@code ChronicleMap} instances don't depend on the builder.
*
* <p>{@code AbstractChronicleMapBuilder} and it's subclasses are mutable, see a note in
* {@link ChronicleHashBuilder} interface documentation.
*
* <p>Later in this documentation, "ChronicleMap" means "ChronicleMaps, created by {@code
* AbstractChronicleMapBuilder}", unless specified different, because theoretically someone might
* provide {@code ChronicleMap} implementations with completely different properties.
*
* <p>{@code ChronicleMap} ("ChronicleMaps, created by {@code AbstractChronicleMapBuilder}")
* currently doesn't support resizing. That is why you should <i>always</i> configure {@linkplain
* #entries(long) number of entries} you are going to insert into the created map <i>at most</i>.
* See {@link #entries(long)} method documentation for more information on this.
*
* <p>{@code ChronicleMap} allocates memory by equally sized chunks. This size is called {@linkplain
* #entrySize(int) entry size}, you are strongly recommended to configure it to achieve least memory
* consumption and best speed. See {@link #entrySize(int)} method documentation for more information
* on this.
*
* @param <K> key type of the maps, produced by this builder
* @param <V> value type of the maps, produced by this builder
* @see ChronicleMap
* @see ChronicleSetBuilder
*/
public abstract class AbstractChronicleMapBuilder<K, V,
B extends AbstractChronicleMapBuilder<K, V, B>>
implements Cloneable, ChronicleHashBuilder<K, ChronicleMap<K, V>, B> {
private static final Bytes EMPTY_BYTES = new ByteBufferBytes(ByteBuffer.allocate(0));
private static final int DEFAULT_KEY_OR_VALUE_SIZE = 120;
private static final int MAX_SEGMENTS = (1 << 30);
private static final int MAX_SEGMENTS_TO_CHAISE_COMPACT_MULTI_MAPS = (1 << 20);
static final short UDP_REPLICATION_MODIFICATION_ITERATOR_ID = 128;
private static final Logger LOG =
LoggerFactory.getLogger(AbstractChronicleMapBuilder.class.getName());
SerializationBuilder<K> keyBuilder;
SerializationBuilder<V> valueBuilder;
private Map<Class<? extends Replicator>, Replicator> replicators = new HashMap<Class<? extends Replicator>, Replicator>();
// used when configuring the number of segments.
private int minSegments = -1;
private int actualSegments = -1;
// used when reading the number of entries per
private long actualEntriesPerSegment = -1L;
private int keySize = 0;
private K sampleKey;
private int valueSize = 0;
private V sampleValue;
private int entrySize = 0;
private Alignment alignment = Alignment.OF_4_BYTES;
private long entries = 1 << 20; // 1 million by default
private long lockTimeOut = 2000;
private TimeUnit lockTimeOutUnit = TimeUnit.MILLISECONDS;
private int metaDataBytes = 0;
private ChronicleHashErrorListener errorListener = ChronicleHashErrorListeners.logging();
private boolean putReturnsNull = false;
private boolean removeReturnsNull = false;
// replication
private TimeProvider timeProvider = TimeProvider.SYSTEM;
private BytesMarshallerFactory bytesMarshallerFactory;
private ObjectSerializer objectSerializer;
private MapEventListener<K, V, ChronicleMap<K, V>> eventListener =
MapEventListeners.nop();
private V defaultValue = null;
private DefaultValueProvider<K, V> defaultValueProvider = NullValueProvider.INSTANCE;
private PrepareValueBytes<K> prepareValueBytes = null;
private byte identifier = -1;
private StatelessBuilder statelessBuilder;
private File file;
AbstractChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
keyBuilder = new SerializationBuilder<K>(keyClass, SerializationBuilder.Role.KEY);
valueBuilder = new SerializationBuilder<V>(valueClass, SerializationBuilder.Role.VALUE);
}
private static long roundUpMapHeaderSize(long headerSize) {
long roundUp = (headerSize + 127L) & ~127L;
if (roundUp - headerSize < 64)
roundUp += 128;
return roundUp;
}
@Override
public B clone() {
try {
@SuppressWarnings("unchecked")
final B result = (B) super.clone();
result.keyBuilder = keyBuilder.clone();
result.valueBuilder = valueBuilder.clone();
return result;
} catch (CloneNotSupportedException e) {
throw new AssertionError(e);
}
}
abstract B self();
/**
* {@inheritDoc}
*
* <p>Example: if keys in your map(s) are English words in {@link String} form, keys size 10
* (a bit more than average English word length) would be a good choice: <pre>{@code
* ChronicleMap<String, LongValue> wordFrequencies = OffHeapUpdatableChronicleMapBuilder
* .of(String.class, LongValue.class)
* .entries(50000)
* .keySize(10)
* // shouldn't specify valueSize(), because it is statically known
* .create();}</pre>
* (Note that 10 is chosen as key size in bytes despite strings in Java are UTF-16 encoded (and
* each character takes 2 bytes on-heap), because default off-heap {@link String} encoding is
* UTF-8 in {@code ChronicleMap}.)
*
* @see #constantKeySizeBySample(Object)
* @see #entrySize(int)
*/
@Override
public B keySize(int keySize) {
if (keySize <= 0)
throw new IllegalArgumentException("Key size must be positive");
this.keySize = keySize;
return self();
}
/**
* {@inheritDoc}
*
* <p>For example, if your keys are Git commit hashes:<pre>{@code
* Map<byte[], String> gitCommitMessagesByHash =
* ChronicleMapBuilder.of(byte[].class, String.class)
* .constantKeySizeBySample(new byte[20])
* .create();}</pre>
*
* @see #keySize(int)
*/
@Override
public B constantKeySizeBySample(K sampleKey) {
this.sampleKey = sampleKey;
return self();
}
private int keySize() {
return keyOrValueSize(keySize, keyBuilder);
}
B valueSize(int valueSize) {
if (valueSize <= 0)
throw new IllegalArgumentException("Value size must be positive");
this.valueSize = valueSize;
return self();
}
B constantValueSizeBySample(V sampleValue) {
this.sampleValue = sampleValue;
return self();
}
int valueSize() {
return keyOrValueSize(valueSize, valueBuilder);
}
private int keyOrValueSize(int configuredSize, SerializationBuilder builder) {
if (configuredSize > 0)
return configuredSize;
// this means size is statically known
if (builder.sizeMarshaller().sizeEncodingSize(0L) == 0)
return (int) builder.sizeMarshaller().readSize(EMPTY_BYTES);
return DEFAULT_KEY_OR_VALUE_SIZE;
}
/**
* {@inheritDoc}
*
* <p>In fully default case you can expect entry size to be about 256 bytes. But it is strongly
* recommended always to configure {@linkplain #keySize(int) key size} and {@linkplain
* #valueSize(int) value size}, if they couldn't be derived statically.
*
* <p>If entry size is not configured explicitly by calling this method, it is computed based on
* {@linkplain #metaDataBytes(int) meta data bytes}, plus {@linkplain #keySize(int) key size},
* plus {@linkplain #valueSize(int) value size}, plus a few bytes required by implementations,
* with respect to {@linkplain #entryAndValueAlignment(Alignment) alignment}.
*
* <p>Note that the actual entrySize will be aligned to 4 (default {@linkplain
* #entryAndValueAlignment(Alignment) entry alignment}). I. e. if you set entry size to 30, the
* actual entry size will be 32 (30 aligned to 4 bytes). If you don't want entry size to be
* aligned, set {@code entryAndValueAlignment(Alignment.NO_ALIGNMENT)}.
*
* @see #entryAndValueAlignment(Alignment)
* @see #entries(long)
*/
@Override
public B entrySize(int entrySize) {
if (entrySize <= 0)
throw new IllegalArgumentException("Entry Size must be positive");
this.entrySize = entrySize;
return self();
}
int entrySize() {
if (entrySize > 0)
return entryAndValueAlignment().alignSize(entrySize);
int size = metaDataBytes;
int keySize = keySize();
size += keyBuilder.sizeMarshaller().sizeEncodingSize(keySize);
size += keySize;
if (useReplication())
size += ReplicatedChronicleMap.ADDITIONAL_ENTRY_BYTES;
int valueSize = valueSize();
size += valueBuilder.sizeMarshaller().sizeEncodingSize(valueSize);
size = entryAndValueAlignment().alignSize(size);
size += valueSize;
// Some cache line heuristics
for (int i = 1; i <= 4; i++) {
int bound = i * 64;
// Not more than 5% oversize.
// DEFAULT_KEY_OR_VALUE_SIZE and this heuristic are specially adjusted to produce
// entry size 256 -- the default value prior to keySize and valueSize -- when key and
// value sizes are both not static nor configured, in both vanilla and replicated modes.
if (size < bound && (bound - size) <= bound / 20) {
size = bound;
break;
}
}
return entryAndValueAlignment().alignSize(size);
}
/**
* Configures alignment strategy of address in memory of entries and independently of address in
* memory of values within entries in ChronicleMaps, created by this builder.
*
* <p>Useful when values of the map are updated intensively, particularly fields with volatile
* access, because it doesn't work well if the value crosses cache lines. Also, on some
* (nowadays rare) architectures any misaligned memory access is more expensive than aligned.
*
* <p>Note that {@linkplain #entrySize(int) entry size} will be aligned according to this
* alignment. I. e. if you set {@code entrySize(20)} and {@link Alignment#OF_8_BYTES}, actual
* entry size will be 24 (20 aligned to 8 bytes).
*
* <p>Default is {@link Alignment#OF_4_BYTES}.
*
* @param alignment the new alignment of the maps constructed by this builder
* @return this {@code ChronicleMapBuilder} back
*/
public B entryAndValueAlignment(Alignment alignment) {
this.alignment = alignment;
return self();
}
Alignment entryAndValueAlignment() {
return alignment;
}
@Override
public B entries(long entries) {
if (entries <= 0L)
throw new IllegalArgumentException("Entries should be positive, " + entries + " given");
this.entries = entries;
return self();
}
long entries() {
return entries;
}
@Override
public B actualEntriesPerSegment(long actualEntriesPerSegment) {
if (actualEntriesPerSegment <= 0L)
throw new IllegalArgumentException("entries per segment should be positive, " +
actualEntriesPerSegment + " given");
if (tooManyEntriesPerSegment(actualEntriesPerSegment))
throw new IllegalArgumentException("max entries per segment is " +
VanillaIntIntMultiMap.MAX_CAPACITY + ", " + actualEntriesPerSegment + " given");
this.actualEntriesPerSegment = actualEntriesPerSegment;
return self();
}
private boolean tooManyEntriesPerSegment(long entriesPerSegment) {
return entriesPerSegment > VanillaIntIntMultiMap.MAX_CAPACITY;
}
long actualEntriesPerSegment() {
if (actualEntriesPerSegment > 0L)
return actualEntriesPerSegment;
int actualSegments = actualSegments();
long actualEntries = totalEntriesIfPoorDistribution(actualSegments);
long actualEntriesPerSegment = divideUpper(actualEntries, actualSegments);
if (tooManyEntriesPerSegment(actualEntriesPerSegment))
throw new IllegalStateException("max entries per segment is " +
VanillaIntIntMultiMap.MAX_CAPACITY + " configured entries() and " +
"actualSegments() so that there should be " + actualEntriesPerSegment +
" entries per segment");
return actualEntriesPerSegment;
}
private long totalEntriesIfPoorDistribution(int segments) {
if (segments == 1)
return entries;
double poorDistEntriesScale = Math.log(segments) * entries;
if (segments <= 8)
return Math.min(entries * segments,
(long) (entries + poorDistEntriesScale * 0.08 + 64)); // 8% was min for tests
return (long) (entries + poorDistEntriesScale * 0.11 + 80); // 11% was min for tests
}
@Override
public B minSegments(int minSegments) {
checkSegments(minSegments);
this.minSegments = minSegments;
return self();
}
int minSegments() {
return Math.max(estimateSegments(), minSegments);
}
private int estimateSegments() {
for (int power = 0; power < 7; power++) {
if (entries < (1L << (7 + (2 * power))))
return 1 << power;
}
// Heuristic -- number of segments ~= cube root from number of entries seems optimal
int maxSegments = 65536;
for (int segments = 4; segments < maxSegments; segments <<= 1) {
if (((long) segments) * segments * segments >= entries)
return segments;
}
return maxSegments;
}
@Override
public B actualSegments(int actualSegments) {
checkSegments(actualSegments);
this.actualSegments = actualSegments;
return self();
}
private static void checkSegments(int segments) {
if (segments <= 0 || segments > MAX_SEGMENTS)
throw new IllegalArgumentException("segments should be positive, " +
segments + " given");
if (segments > MAX_SEGMENTS)
throw new IllegalArgumentException("Max segments is " + MAX_SEGMENTS + ", " +
segments + " given");
}
int actualSegments() {
if (actualSegments > 0)
return actualSegments;
long shortMMapSegments = trySegments(VanillaShortShortMultiMap.MAX_CAPACITY,
MAX_SEGMENTS_TO_CHAISE_COMPACT_MULTI_MAPS);
if (shortMMapSegments > 0L)
return (int) shortMMapSegments;
long intMMapSegments = trySegments(VanillaIntIntMultiMap.MAX_CAPACITY, MAX_SEGMENTS);
if (intMMapSegments > 0L)
return (int) intMMapSegments;
throw new IllegalStateException("Max segments is " + MAX_SEGMENTS + ", configured so much" +
" entries() that builder automatically decided to use " +
(-intMMapSegments) + " segments");
}
private long trySegments(long maxSegmentCapacity, int maxSegments) {
long segments = divideUpper(totalEntriesIfPoorDistribution(minSegments()),
maxSegmentCapacity);
segments = Maths.nextPower2(Math.max(segments, minSegments()), 1L);
return segments <= maxSegments ? segments : -segments;
}
private static long divideUpper(long dividend, long divisor) {
return ((dividend - 1L) / divisor) + 1L;
}
private boolean canSupportShortShort() {
return entries > (long) minSegments() << 15;
}
public B lockTimeOut(long lockTimeOut, TimeUnit unit) {
this.lockTimeOut = lockTimeOut;
lockTimeOutUnit = unit;
return self();
}
long lockTimeOut(TimeUnit unit) {
return unit.convert(lockTimeOut, lockTimeOutUnit);
}
@Override
public B errorListener(ChronicleHashErrorListener errorListener) {
this.errorListener = errorListener;
return self();
}
ChronicleHashErrorListener errorListener() {
return errorListener;
}
/**
* Configures if the maps created by this builder should return {@code null} instead of previous
* mapped values on {@link ChronicleMap#put(Object, Object) ChornicleMap.put(key, value)}
* calls.
*
* <p>{@link Map#put(Object, Object) Map.put()} returns the previous value, functionality which
* is rarely used but fairly cheap for {@link HashMap}. In the case, for an off heap collection,
* it has to create a new object and deserialize the data from off-heap memory. It's expensive
* for something you probably don't use.
*
* <p>By default, of cause, {@code ChronicleMap} conforms the general {@code Map} contract and
* returns the previous mapped value on {@code put()} calls.
*
* @param putReturnsNull {@code true} if you want {@link ChronicleMap#put(Object, Object)
* ChronicleMap.put()} to not return the value that was replaced but
* instead return {@code null}
* @return an instance of the map builder back
* @see #removeReturnsNull(boolean)
*/
public B putReturnsNull(boolean putReturnsNull) {
this.putReturnsNull = putReturnsNull;
return self();
}
boolean putReturnsNull() {
return putReturnsNull;
}
/**
* Configures if the maps created by this builder should return {@code null} instead of the last
* mapped value on {@link ChronicleMap#remove(Object) ChronicleMap.remove(key)} calls.
*
* <p>{@link Map#remove(Object) Map.remove()} returns the previous value, functionality which is
* rarely used but fairly cheap for {@link HashMap}. In the case, for an off heap collection, it
* has to create a new object and deserialize the data from off-heap memory. It's expensive for
* something you probably don't use.
*
* @param removeReturnsNull {@code true} if you want {@link ChronicleMap#remove(Object)
* ChronicleMap.remove()} to not return the value of the removed entry
* but instead return {@code null}
* @return an instance of the map builder back
* @see #putReturnsNull(boolean)
*/
public B removeReturnsNull(boolean removeReturnsNull) {
this.removeReturnsNull = removeReturnsNull;
return self();
}
boolean removeReturnsNull() {
return removeReturnsNull;
}
@Override
public B metaDataBytes(int metaDataBytes) {
if ((metaDataBytes & 0xFF) != metaDataBytes)
throw new IllegalArgumentException("MetaDataBytes must be [0..255] was " + metaDataBytes);
this.metaDataBytes = metaDataBytes;
return self();
}
int metaDataBytes() {
return metaDataBytes;
}
@Override
public String toString() {
return "ChronicleMapBuilder{" +
"actualSegments=" + pretty(actualSegments) +
", minSegments=" + pretty(minSegments) +
", actualEntriesPerSegment=" + pretty(actualEntriesPerSegment) +
", keySize=" + pretty(keySize) +
", sampleKeyForConstantSizeComputation=" + pretty(sampleKey) +
", valueSize=" + pretty(valueSize) +
", sampleValueForConstantSizeComputation=" + pretty(sampleValue) +
", entrySize=" + pretty(entrySize) +
", entryAndValueAlignment=" + entryAndValueAlignment() +
", entries=" + entries() +
", lockTimeOut=" + lockTimeOut + " " + lockTimeOutUnit +
", metaDataBytes=" + metaDataBytes() +
", errorListener=" + errorListener() +
", putReturnsNull=" + putReturnsNull() +
", removeReturnsNull=" + removeReturnsNull() +
", timeProvider=" + timeProvider() +
", bytesMarshallerFactory=" + pretty(bytesMarshallerFactory) +
", objectSerializer=" + pretty(objectSerializer) +
", keyBuilder=" + keyBuilder +
", valueBuilder=" + valueBuilder +
", eventListener=" + eventListener +
", defaultValue=" + defaultValue +
", defaultValueProvider=" + pretty(defaultValueProvider) +
", prepareValueBytes=" + pretty(prepareValueBytes) +
'}';
}
private static String pretty(int value) {
return value > 0 ? value + "" : "not configured";
}
private static String pretty(Object obj) {
return obj != null ? obj + "" : "not configured";
}
@SuppressWarnings("EqualsWhichDoesntCheckParameterClass")
@Override
public boolean equals(Object o) {
return builderEquals(this, o);
}
@Override
public int hashCode() {
return toString().hashCode();
}
public B replicators(byte identifier, ReplicationConfig... replicationConfigs) {
this.identifier = identifier;
this.replicators.clear();
for (ReplicationConfig replicationConfig : replicationConfigs) {
Replicator replicator;
if (replicationConfig instanceof TcpReplicationConfig) {
replicator = Replicators.tcp((TcpReplicationConfig) replicationConfig);
} else if (replicationConfig instanceof UdpReplicationConfig) {
replicator = Replicators.udp((UdpReplicationConfig) replicationConfig);
} else
throw new UnsupportedOperationException();
this.replicators.put(replicator.getClass(), replicator);
}
return self();
}
public B channel(ChannelProvider.ChronicleChannel chronicleChannel) {
this.identifier = chronicleChannel.identifier();
this.replicators.clear();
replicators.put(chronicleChannel.getClass(), chronicleChannel);
return self();
}
@Override
public B disableReplication() {
identifier = -1;
replicators.clear();
return self();
}
@Override
public B timeProvider(TimeProvider timeProvider) {
this.timeProvider = timeProvider;
return self();
}
TimeProvider timeProvider() {
return timeProvider;
}
BytesMarshallerFactory bytesMarshallerFactory() {
return bytesMarshallerFactory == null ?
bytesMarshallerFactory = new VanillaBytesMarshallerFactory() :
bytesMarshallerFactory;
}
@Override
public B bytesMarshallerFactory(BytesMarshallerFactory bytesMarshallerFactory) {
this.bytesMarshallerFactory = bytesMarshallerFactory;
return self();
}
ObjectSerializer objectSerializer() {
return objectSerializer == null ?
objectSerializer = BytesMarshallableSerializer.create(
bytesMarshallerFactory(), JDKObjectSerializer.INSTANCE) :
objectSerializer;
}
@Override
public B objectSerializer(ObjectSerializer objectSerializer) {
this.objectSerializer = objectSerializer;
return self();
}
/**
* {@inheritDoc}
*
* <p>Example:
*/
@Override
public B keyMarshaller(@NotNull BytesMarshaller<K> keyMarshaller) {
keyBuilder.marshaller(keyMarshaller, null);
return self();
}
@Override
public B keyDeserializationFactory(@NotNull ObjectFactory<K> keyDeserializationFactory) {
keyBuilder.factory(keyDeserializationFactory);
return self();
}
@Override
public B immutableKeys() {
keyBuilder.instancesAreMutable(false);
return self();
}
@Override
public B stateless(@NotNull StatelessBuilder statelessBuilder) {
this.statelessBuilder = statelessBuilder;
return self();
}
@Override
public B file(File file) throws IOException {
this.file = file;
return self();
}
/**
* Configures factory which is used to create a new value instance, if value class is either
* {@link Byteable}, {@link BytesMarshallable} or {@link Externalizable} subclass in maps,
* created by this builder.
*
* <p>Default value deserialization factory is {@link NewInstanceObjectFactory}, which creates
* a new value instance using {@link Class#newInstance()} default constructor. You could provide
* an {@link AllocateInstanceObjectFactory}, which uses {@code Unsafe.allocateInstance(Class)}
* (you might want to do this for better performance or if you don't want to initialize fields),
* or a factory which calls a value class constructor with some arguments, or a factory which
* internally delegates to instance pool or {@link ThreadLocal}, to reduce allocations.
*
* @param valueDeserializationFactory the value factory used to produce instances to deserialize
* data in
* @return this builder back
*/
public B valueDeserializationFactory(@NotNull ObjectFactory<V> valueDeserializationFactory) {
valueBuilder.factory(valueDeserializationFactory);
return self();
}
public B eventListener(MapEventListener<K, V, ChronicleMap<K, V>> eventListener) {
this.eventListener = eventListener;
return self();
}
MapEventListener<K, V, ChronicleMap<K, V>> eventListener() {
return eventListener;
}
/**
* Specifies the value to be put for each key queried in {@link ChronicleMap#acquireUsing
* acquireUsing()} method, if the key is absent in the map, created by this builder.
*
* <p>This configuration overrides any previous {@link #defaultValueProvider(
* DefaultValueProvider)} configuration to this {@code AbstractChronicleMapBuilder}.
*
* @param defaultValue the default value to be put to the map for absent keys during {@code
* acquireUsing()} calls
* @return this builder object back
*/
public B defaultValue(V defaultValue) {
this.defaultValue = defaultValue;
this.defaultValueProvider = null;
if (defaultValue == null)
defaultValueProvider = NullValueProvider.INSTANCE;
this.prepareValueBytes = null;
return self();
}
/**
* Specifies the function to obtain a value for the key during {@link ChronicleMap#acquireUsing
* acquireUsing()} calls, if the key is absent in the map, created by this builder.
*
* <p>This configuration overrides any previous {@link #defaultValue(Object)} configuration to
* this {@code AbstractChronicleMapBuilder}.
*
* @param defaultValueProvider the strategy to obtain a default value by the absent key
* @return this builder object back
*/
public B defaultValueProvider(
@NotNull DefaultValueProvider<K, V> defaultValueProvider) {
this.defaultValueProvider = defaultValueProvider;
this.defaultValue = null;
this.prepareValueBytes = null;
return self();
}
B prepareValueBytesOnAcquire(@NotNull PrepareValueBytes<K> prepareValueBytes) {
this.prepareValueBytes = prepareValueBytes;
this.defaultValue = null;
this.defaultValueProvider = null;
return self();
}
PrepareValueBytesAsWriter<K> prepareValueBytesAsWriter() {
return new PrepareValueBytesAsWriter<>(prepareValueBytes, valueSize());
}
/**
* Non-public because should be called only after {@link #preMapConstruction()}
*/
DefaultValueProvider<K, V> defaultValueProvider() {
if (defaultValueProvider != null)
return defaultValueProvider;
if (defaultValue == null)
return null;
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> createWithFile(File file) throws IOException {
for (int i = 0; i < 10; i++) {
if (file.exists() && file.length() > 0) {
FileInputStream fis = new FileInputStream(file);
ObjectInputStream ois = new ObjectInputStream(fis);
try {
VanillaChronicleMap<K, ?, ?, V, ?, ?> map =
(VanillaChronicleMap<K, ?, ?, V, ?, ?>) ois.readObject();
map.headerSize = roundUpMapHeaderSize(fis.getChannel().position());
map.createMappedStoreAndSegments(file);
return establishReplication(map);
} catch (ClassNotFoundException e) {
throw new IOException(e);
} finally {
ois.close();
}
}
if (file.createNewFile() || file.length() == 0) {
break;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
throw new IOException(e);
}
}
// new file
if (!file.exists())
throw new FileNotFoundException("Unable to create " + file);
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap();
FileOutputStream fos = new FileOutputStream(file);
ObjectOutputStream oos = new ObjectOutputStream(fos);
try {
oos.writeObject(map);
oos.flush();
map.headerSize = roundUpMapHeaderSize(fos.getChannel().position());
map.createMappedStoreAndSegments(file);
} finally {
oos.close();
}
return establishReplication(map);
}
@Override
public ChronicleMap<K, V> create() throws IOException {
if (statelessBuilder != null)
return createStatelessMap(statelessBuilder);
return (file != null) ? createWithFile(file) : createWithoutFile();
}
private ChronicleMap<K, V> createWithoutFile() throws IOException {
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap();
BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE,
map.sizeInBytes(), true);
map.createMappedStoreAndSegments(bytesStore);
return establishReplication(map);
}
private ChronicleMap<K, V> createStatelessMap(StatelessBuilder statelessBuilder) throws IOException {
preMapConstruction();
final KeyValueSerializer<K, V> keyValueSerializer
= new KeyValueSerializer<K, V>(keyBuilder, valueBuilder);
final Class<K> kClass = keyBuilder.eClass;
final Class<V> vClass = valueBuilder.eClass;
return new StatelessChronicleMap<K, V>(
keyValueSerializer,
statelessBuilder,
entrySize(),
kClass,
vClass);
}
private VanillaChronicleMap<K, ?, ?, V, ?, ?> newMap() throws IOException {
preMapConstruction();
if (useReplication()) {
return new ReplicatedChronicleMap<K, Object, MetaBytesInterop<K, Object>,
V, Object, MetaBytesWriter<V, Object>>(this);
} else {
return new VanillaChronicleMap<K, Object, MetaBytesInterop<K, Object>,
V, Object, MetaBytesWriter<V, Object>>(this);
}
}
private boolean useReplication() {
return identifier != -1;
}
void preMapConstruction() {
keyBuilder.objectSerializer(objectSerializer());
valueBuilder.objectSerializer(objectSerializer());
long maxSize = (long) entrySize() * figureBufferAllocationFactor();
keyBuilder.maxSize(maxSize);
valueBuilder.maxSize(maxSize);
if (sampleKey != null)
keyBuilder.constantSizeBySample(sampleKey);
if (sampleValue != null)
valueBuilder.constantSizeBySample(sampleValue);
}
private ChronicleMap<K, V> establishReplication(ChronicleMap<K, V> map)
throws IOException {
if (map instanceof ReplicatedChronicleMap) {
ReplicatedChronicleMap result = (ReplicatedChronicleMap) map;
for (Replicator replicator : replicators.values()) {
Closeable token = replicator.applyTo(this, result, result, map);
if (replicators.size() == 1 && token.getClass() == UdpReplicator.class) {
LOG.warn(
"MISSING TCP REPLICATION : The UdpReplicator only attempts to read data " +
"(it does not enforce or guarantee delivery), you should use" +
"the UdpReplicator if you have a large number of nodes, and you wish" +
"to receive the data before it becomes available on TCP/IP. Since data" +
"delivery is not guaranteed, it is recommended that you only use" +
"the UDP Replicator in conjunction with a TCP Replicator"
);
}
result.addCloseable(token);
}
}
return map;
}
private int figureBufferAllocationFactor() {
// if expected map size is about 1000, seems rather wasteful to allocate
// key and value serialization buffers each x64 of expected entry size..
return (int) Math.min(Math.max(2L, entries() >> 10),
VanillaChronicleMap.MAX_ENTRY_OVERSIZE_FACTOR);
}
public byte identifier() {
return identifier;
}
}
@@ -23,7 +23,7 @@ import net.openhft.lang.io.Bytes;
/**
* Memory addresses alignment strategies.
*
* @see ChronicleMapBuilder#entryAndValueAlignment(Alignment)
* @see AbstractChronicleMapBuilder#entryAndValueAlignment(Alignment)
*/
public enum Alignment {
/**
@@ -409,7 +409,7 @@ public final class ChannelProvider implements Closeable {
}
@Override
protected Closeable applyTo(ChronicleMapBuilder builder,
protected Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica map, EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap) {
if (builder.entrySize() > maxEntrySize) {
throw new IllegalArgumentException("During ReplicatingChannelBuilder setup, " +
@@ -22,7 +22,6 @@ import net.openhft.chronicle.common.ChronicleHash;
import net.openhft.chronicle.common.serialization.BytesReader;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.serialization.BytesMarshaller;
import net.openhft.lang.io.serialization.ObjectFactory;
import net.openhft.lang.model.Byteable;
import java.io.Closeable;
@@ -38,11 +37,15 @@ import java.util.concurrent.ConcurrentMap;
* <li>{@code ChronicleMap} flavors and properties</li>
* <li>available configurations</li>
* </ul>
* see {@link ChronicleMapBuilder} documentation.
* see {@link AbstractChronicleMapBuilder} documentation.
*
* <p>Functionally this interface defines some methods supporting garbage-free off-heap programming:
* {@link #getUsing(Object, Object)}, {@link #acquireUsing(Object, Object)}.
*
* <p>Roughly speaking, {@code ChronicleMap} compares keys and values by their binary serialized
* form, that shouldn't necessary be the same equality relation as defined by built-in {@link
* Object#equals(Object)} method, which is prescribed by general {@link Map} contract.
*
* <p>Note that {@code ChronicleMap} extends {@link Closeable}, don't forget
* to {@linkplain #close() close} map when it is no longer needed.
*
@@ -62,15 +65,6 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
* Returns the value to which the specified key is mapped, or {@code null} if this map contains
* no mapping for the key.
*
* <p>If the specified key if absent in the map, {@linkplain
* ChronicleMapBuilder#defaultValue(Object) default value} is taken or {@linkplain
* ChronicleMapBuilder#defaultValueProvider(DefaultValueProvider) default value provider}
* is called (depending on what was configured last in {@code ChronicleMapBuilder}).
* If it results to something different from {@code null}, this object is put to this map for
* the specified key, and then returned from this {@code get()} call. Otherwise, if default
* value is {@code null} and default value provider returns {@code null}, the map remains
* unchanged and {@code null} is returned.
*
* <p>If the value class allows reusing, particularly if it is a {@link Byteable} subclass,
* consider {@link #getUsing(Object, Object)} method instead of this to reduce garbage creation.
*
@@ -90,40 +84,59 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
* {@code value} object via value marshaller's {@link BytesMarshaller#read(Bytes, Object)
* read(Bytes, value)} or value reader's {@link BytesReader#read(Bytes, long, Object)
* read(Bytes, size, value)} method, depending on what deserialization strategy is configured
* on this map builder. If the value deserializer is able to reuse the given {@code value}
* object, calling this method instead of {@link #get(Object)} could help to reduce garbage
* creation.
* on the builder, using which this map was constructed. If the value deserializer is able
* to reuse the given {@code value} object, calling this method instead of {@link #get(Object)}
* could help to reduce garbage creation.
*
* <p>The provided {@code value} object is allowed to be {@code null}, in this case
* {@code map.getUsing(key, null)} call is semantically equivalent to simple
* {@code map.get(key)} call.
*
* <p>If the specified key if absent in the map, {@linkplain
* ChronicleMapBuilder#defaultValue(Object) default value} is taken or {@linkplain
* ChronicleMapBuilder#defaultValueProvider(DefaultValueProvider) default value provider}
* is called. If it results to something different from {@code null}, this object is put
* to this map for the specified key, and then a new value is returned from this
* {@code getUsing()} call. Otherwise, if default value is {@code null} and default value
* provider, if specified, returns {@code null}, the map remains unchanged and {@code null}
* is returned. The provided {@code value} is untouched anyway in this case.
*
* @param key the key whose associated value is to be returned
* @param value the object to read value data in, if possible
* @return the value mapped to the specified key after this method call, or {@code null} if no
* value is mapped
* @param key the key whose associated value is to be returned
* @param usingValue the object to read value data in, if possible
* @return the value to which the specified key is mapped, or {@code null} if this map contains
* no mapping for the key
* @see #get(Object)
* @see ChronicleMapBuilder#valueMarshallerAndFactory(BytesMarshaller, ObjectFactory)
* @see #acquireUsing(Object, Object)
* @see ChronicleMapBuilder#valueMarshaller(BytesMarshaller)
*/
V getUsing(K key, V value);
V getUsing(K key, V usingValue);
/**
* Acquire a value for a key, creating if absent. If the value is Byteable, it will be assigned
* to reference the value, instead of copying the data.
* Acquire a value for a key, creating if absent.
*
* @param key the key whose associated value is to be returned
* @param value to reuse if possible. If null, a new object will be created.
* @return value created or found.
* <p>If the specified key if absent in the map, {@linkplain
* AbstractChronicleMapBuilder#defaultValue(Object) default value} is taken or {@linkplain
* AbstractChronicleMapBuilder#defaultValueProvider(DefaultValueProvider) default value
* provider} is called. Then this object is put to this map for the specified key.
*
* <p>Then, either if the key was initially absent in the map or already present,
* the value is deserialized just as during {@link #getUsing(Object, Object)
* getUsing(key, usingValue)} call, passed the same {@code key} and {@code usingValue} as into
* this method call. This means, as in {@link #getUsing}, {@code usingValue} could safely be
* {@code null}, in this case a new value instance is created to deserialize the data.
*
* <p>In code, {@code acquireUsing} is specified as <pre>{@code
* V acquireUsing(K key, V usingValue) {
* if (!containsKey(key))
* put(key, defaultValue(key));
* return getUsing(key, usingValue);
* }}</pre> Where {@code defaultValue(key)} returns either {@linkplain
* AbstractChronicleMapBuilder#defaultValue(Object) default value} or {@link
* AbstractChronicleMapBuilder#defaultValueProvider(DefaultValueProvider) defaultValueProvider.}
*
* <p>If the {@code ChronicleMap} is off-heap updatable, i. e. created via {@link
* OffHeapUpdatableChronicleMapBuilder} builder (values are {@link Byteable}), there is one more
* option of what to do if the key is absent in the map, see {@link
* OffHeapUpdatableChronicleMapBuilder#prepareValueBytesOnAcquire(PrepareValueBytes)}.
* By default, value bytes are just zeroed out, no default value, either provided for key
* or constant, is put for the absent key.
*
* @param key the key whose associated value is to be returned
* @param usingValue the object to read value data in, if possible
* @return value to which the given key is mapping after this call, either found or created
* @see #getUsing(Object, Object)
*/
V acquireUsing(K key, V value);
V acquireUsing(K key, V usingValue);
}
File diff suppressed because it is too large Load Diff
@@ -48,7 +48,7 @@ final class ConstantValueProvider<K, V> implements DefaultValueProvider<K, V> {
}
@Override
public V get(K key, V usingValue) {
public V get(K key) {
return value;
}
}
@@ -22,26 +22,19 @@ import java.io.Serializable;
/**
* Default value computation strategy, used
* in {@link ChronicleMapBuilder#defaultValueProvider(DefaultValueProvider)} configuration.
* in {@link AbstractChronicleMapBuilder#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.
* Returns a value to be put during {@link ChronicleMap#acquireUsing(Object, Object)} call
* for the specified key, if it is absent 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);
V get(K key);
}
@@ -33,7 +33,7 @@ import java.io.Serializable;
* class and override only methods corresponding the events you are interested in.
*
* <p>To configure {@code MapEventListener} for {@code ChronicleMap}, use
* {@link ChronicleMapBuilder#eventListener(MapEventListener)} method.
* {@link AbstractChronicleMapBuilder#eventListener(MapEventListener)} method.
*
* <p>{@link MapEventListeners} uninstantiable class contains several logging implementations and
* the default {@linkplain MapEventListeners#nop() no-op implementation}.
@@ -44,7 +44,7 @@ import java.io.Serializable;
* @param <K> key type of the maps, trackable by this event listener
* @param <V> value type of the maps, trackable by this event listener
* @param <M> {@code ChronicleMap} subtype, trackable by this event listener
* @see ChronicleMapBuilder#eventListener(MapEventListener)
* @see AbstractChronicleMapBuilder#eventListener(MapEventListener)
* @see MapEventListeners
*/
public abstract class MapEventListener<K, V, M extends ChronicleMap<K, V>>
@@ -75,7 +75,7 @@ public abstract class MapEventListener<K, V, M extends ChronicleMap<K, V>>
* This method is call whenever a new value is put for the key in the map during calls of such
* methods as {@link ChronicleMap#put put}, {@link ChronicleMap#putIfAbsent putIfAbsent},
* {@link ChronicleMap#replace(Object, Object, Object) replace}, etc. When a new value is
* {@linkplain ChronicleMapBuilder#defaultValue(Object) default} for the map or obtained during
* {@linkplain AbstractChronicleMapBuilder#defaultValue(Object) default} for the map or obtained during
* {@link ChronicleMap#acquireUsing acquireUsing} call is put for the key, this method is called
* as well.
*
@@ -22,7 +22,7 @@ enum NullValueProvider implements DefaultValueProvider {
INSTANCE;
@Override
public Object get(Object key, Object usingValue) {
public Object get(Object key) {
return null;
}
}
@@ -0,0 +1,98 @@
/*
* 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 org.jetbrains.annotations.NotNull;
public final class OffHeapUpdatableChronicleMapBuilder<K, V>
extends AbstractChronicleMapBuilder<K, V, OffHeapUpdatableChronicleMapBuilder<K, V>> {
public static <K, V> OffHeapUpdatableChronicleMapBuilder<K, V> of(
@NotNull Class<K> keyClass, @NotNull Class<V> valueClass) {
if (!Byteable.class.isAssignableFrom(valueClass)) {
if (!valueClass.isInterface()) {
throw new IllegalArgumentException(
"Value class should be either Byteable subclass or interface," +
"allowing direct Byteable implementation generation");
}
valueClass = DataValueClasses.directClassFor(valueClass);
}
return new OffHeapUpdatableChronicleMapBuilder<K, V>(keyClass, valueClass);
}
OffHeapUpdatableChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
super(keyClass, valueClass);
prepareValueBytesOnAcquire(new ZeroOutValueBytes<K>(valueSize()));
}
@Override
OffHeapUpdatableChronicleMapBuilder<K, V> self() {
return this;
}
/**
* {@inheritDoc} Also, it overrides any previous {@link #prepareValueBytesOnAcquire}
* configuration to this {@code OffHeapUpdatableChronicleMapBuilder}.
*
* <p>By default, the default value is not specified, default {@linkplain
* #prepareValueBytesOnAcquire prepare value bytes routine} is specified instead.
*
* @see #defaultValueProvider(DefaultValueProvider)
* @see #prepareValueBytesOnAcquire(PrepareValueBytes)
*/
@Override
public OffHeapUpdatableChronicleMapBuilder<K, V> defaultValue(V defaultValue) {
return super.defaultValue(defaultValue);
}
/**
* {@inheritDoc} Also, it overrides any previous {@link #prepareValueBytesOnAcquire}
* configuration to this {@code OffHeapUpdatableChronicleMapBuilder}.
*
* <p>By default, the default value provider is not specified, default {@linkplain
* #prepareValueBytesOnAcquire prepare value bytes routine} is specified instead.
*
* @see #defaultValue(Object)
* @see #prepareValueBytesOnAcquire(PrepareValueBytes)
*/
@Override
public OffHeapUpdatableChronicleMapBuilder<K, V> defaultValueProvider(
@NotNull DefaultValueProvider<K, V> defaultValueProvider) {
return super.defaultValueProvider(defaultValueProvider);
}
/**
* Configures the procedure which is called on the bytes, which later the returned value is
* pointing to, if the key is absent, on {@link ChronicleMap#acquireUsing(Object, Object)
* acquireUsing()} call on maps, created by this builder.
*
* <p>The default preparation callback zeroes out the value bytes.
*
* @param prepareValueBytes what to do with the value bytes before assigning them into the
* {@link Byteable} value to return from {@code acquireUsing()} call
* @return this builder back
* @see #defaultValue(Object)
* @see #defaultValueProvider(DefaultValueProvider)
*/
@Override
OffHeapUpdatableChronicleMapBuilder<K, V> prepareValueBytesOnAcquire(
@NotNull PrepareValueBytes<K> prepareValueBytes) {
return super.prepareValueBytesOnAcquire(prepareValueBytes);
}
}
@@ -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;
import net.openhft.lang.io.Bytes;
import java.io.Serializable;
public interface PrepareValueBytes<K> extends Serializable {
void prepare(Bytes bytes, K key);
}
@@ -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.chronicle.common.serialization.BytesWriter;
import net.openhft.chronicle.common.serialization.MetaBytesWriter;
import net.openhft.lang.io.Bytes;
import java.io.Serializable;
final class PrepareValueBytesAsWriter<K>
implements BytesWriter<K>, MetaBytesWriter<K, PrepareValueBytesAsWriter<K>>,
Serializable {
private static final long serialVersionUID = 0L;
private final PrepareValueBytes<K> prepareValueBytes;
private final long valueSize;
PrepareValueBytesAsWriter(PrepareValueBytes<K> prepareValueBytes, long valueSize) {
this.prepareValueBytes = prepareValueBytes;
this.valueSize = valueSize;
}
@Override
public long size(K k) {
return valueSize;
}
@Override
public void write(Bytes bytes, K k) {
prepareValueBytes.prepare(bytes, k);
}
@Override
public long size(PrepareValueBytesAsWriter<K> writer, K k) {
return size(k);
}
@Override
public void write(PrepareValueBytesAsWriter<K> writer, Bytes bytes, K k) {
write(bytes, k);
}
}
@@ -106,7 +106,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
private transient ModificationDelegator modificationDelegator;
private transient long startOfModificationIterators;
public ReplicatedChronicleMap(@NotNull ChronicleMapBuilder<K, V> builder)
public ReplicatedChronicleMap(@NotNull AbstractChronicleMapBuilder<K, V, ?> builder)
throws IOException {
super(builder);
this.timeProvider = builder.timeProvider();
@@ -613,38 +613,39 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
/**
* @see VanillaChronicleMap.Segment#acquire
*/
V acquire(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter,
V acquire(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop,
K key, V usingValue, long hash2, boolean create, long timestamp) {
if (create)
writeLock();
else
readLock();
try {
long keySize = metaKeyWriter.size(keyWriter, key);
long keySize = metaKeyInterop.size(keyInterop, key);
MultiStoreBytes entry = acquireTmpBytes();
long offset = searchKey(keyWriter, metaKeyWriter, key, keySize, hash2, entry,
long offset = searchKey(keyInterop, metaKeyInterop, key, keySize, hash2, entry,
hashLookupLiveOnly);
if (offset >= 0) {
// skip the timestamp, identifier and is deleted flag
entry.skip(ADDITIONAL_ENTRY_BYTES);
if (offset >= 0L) {
return onKeyPresentOnAcquire(copies, key, usingValue, offset, entry);
} else {
boolean usingValuePassed = usingValue != null;
usingValue = tryObtainUsingValueOnAcquire(key, usingValue, create);
if (usingValue != null) {
// see VanillaChronicleMap.Segment.acquire() for explanation
// why `usingValue` is `create`.
offset = putEntryOnAcquire(copies, metaKeyWriter, keyWriter, key, keySize,
hash2, usingValue, create, timestamp);
incrementSize();
if (usingValuePassed || !create)
notifyPut(offset, true, key, usingValue, posFromOffset(offset));
return usingValue;
} else {
if (!create)
return null;
if (defaultValueProvider != null) {
V defaultValue = defaultValueProvider.get(key);
copies = valueWriterProvider.getCopies(copies);
VW valueWriter = valueWriterProvider.get(copies, originalValueWriter);
copies = metaValueWriterProvider.getCopies(copies);
MetaBytesWriter<V, VW> metaValueWriter = metaValueWriterProvider.get(
copies, originalMetaValueWriter, valueWriter, defaultValue);
offset = putEntry(metaKeyInterop, keyInterop, key, keySize, hash2,
localIdentifier, timestamp, hashLookupLiveOnly,
metaValueWriter, valueWriter, defaultValue);
} else {
offset = putEntry(metaKeyInterop, keyInterop, key, keySize, hash2,
localIdentifier, timestamp, hashLookupLiveOnly,
prepareValueBytesAsWriter, null, key);
}
return onKeyAbsentOnAcquire(copies, key, keySize, usingValue, offset);
}
} finally {
if (create)
@@ -654,15 +655,11 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
private long putEntryOnAcquire(ThreadLocalCopies copies,
MKI metaKeyWriter, KI keyWriter, K key,
long keySize, long hash2, V value, boolean usingValue,
long timestamp) {
return putEntry(copies, metaKeyWriter, keyWriter, key, keySize, hash2, value,
usingValue, localIdentifier, timestamp, hashLookupLiveOnly);
@Override
void skipReplicationBytes(Bytes bytes) {
bytes.skip(ADDITIONAL_ENTRY_BYTES);
}
/**
* called from a remote node as part of replication
*/
@@ -858,7 +855,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
// key is not found
long offset = putEntry(copies, metaKeyWriter, keyWriter, key, keySize, hash2, value,
false, identifier, timestamp, hashLookup);
identifier, timestamp, hashLookup);
incrementSize();
notifyPut(offset, true, key, value, posFromOffset(offset));
return null;
@@ -905,9 +902,6 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @param hash2 a hash of the {@code keyBytes}. Caller was searching for the
* key in the {@code searchedHashLookup} using this hash.
* @param value the value to put
* @param usingValue {@code true} if the value should be backed with the bytes of
* the entry, if it implements {@link Byteable} interface, {@code
* false} if it should put itself
* @param identifier the identifier of the outer CHM node
* @param timestamp the timestamp when the entry was put <s>(this could be later if
* it was a remote put)</s> this method is called only from usual
@@ -916,30 +910,25 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* @return offset of the written entry in the Segment bytes
* @see VanillaChronicleMap.Segment#putEntry
*/
private long putEntry(ThreadLocalCopies copies, MKI metaKeyWriter,
KI keyWriter, K key, long keySize, long hash2, V value,
boolean usingValue, final int identifier, final long timestamp,
MultiMap searchedHashLookup) {
private long putEntry(ThreadLocalCopies copies,
MKI metaKeyInterop, KI keyInterop, K key, long keySize, long hash2,
V value,
int identifier, long timestamp, MultiMap searchedHashLookup) {
copies = valueWriterProvider.getCopies(copies);
VW valueWriter = valueWriterProvider.get(copies, originalValueWriter);
copies = metaValueWriterProvider.getCopies(copies);
MetaBytesWriter<V, VW> metaValueWriter = metaValueWriterProvider.get(
copies, originalMetaValueWriter, valueWriter, value);
// "if-else polymorphism" is not very beautiful, but allows to
// reuse the rest code of this method and doesn't hurt performance.
boolean byteableValue = usingValue && value instanceof Byteable;
long valueSize;
MetaBytesWriter<V, VW> metaValueWriter = null;
VW valueWriter = null;
Byteable valueAsByteable = null;
if (!byteableValue) {
copies = valueWriterProvider.getCopies(copies);
valueWriter = valueWriterProvider.get(copies, originalValueWriter);
copies = metaValueWriterProvider.getCopies(copies);
metaValueWriter = metaValueWriterProvider.get(
copies, originalMetaValueWriter, valueWriter, value);
valueSize = metaValueWriter.size(valueWriter, value);
} else {
valueAsByteable = (Byteable) value;
valueSize = valueAsByteable.maxSize();
}
return putEntry(metaKeyInterop, keyInterop, key, keySize, hash2,
identifier, timestamp, searchedHashLookup, metaValueWriter, valueWriter, value);
}
private <E, EW> long putEntry(
MKI metaKeyInterop, KI keyInterop, K key, long keySize, long hash2,
final int identifier, final long timestamp, MultiMap searchedHashLookup,
MetaBytesWriter<E, EW> metaElemWriter, EW elemWriter, E elem) {
long valueSize = metaElemWriter.size(elemWriter, elem);
long entrySize = entrySize(keySize, valueSize);
long pos = alloc(inBlocks(entrySize));
long offset = offsetFromPos(pos);
@@ -947,14 +936,15 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
NativeBytes entry = entry(offset);
keySizeMarshaller.writeSize(entry, keySize);
metaKeyWriter.write(keyWriter, entry, key);
metaKeyInterop.write(keyInterop, entry, key);
entry.writeLong(timestamp);
entry.writeByte(identifier);
entry.writeBoolean(false);
writeValueOnPutEntry(valueSize, metaValueWriter, valueWriter, value, valueAsByteable,
entry);
valueSizeMarshaller.writeSize(entry, valueSize);
alignment.alignPositionAddr(entry);
metaElemWriter.write(elemWriter, entry, elem);
// we have to add it both to the live
if (searchedHashLookup == hashLookupLiveAndDeleted) {
@@ -1573,7 +1563,7 @@ abstract class Replicator {
* already been applied to a map (or the specified number of
* maps)
*/
protected abstract Closeable applyTo(ChronicleMapBuilder builder,
protected abstract Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica map, Replica.EntryExternalizable entryExternalizable,
final ChronicleMap chronicleMap)
throws IOException;
@@ -1589,7 +1579,7 @@ final class Replicators {
return new Replicator() {
@Override
protected Closeable applyTo(ChronicleMapBuilder builder,
protected Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica replica, Replica.EntryExternalizable entryExternalizable,
final ChronicleMap chronicleMap) throws IOException {
@@ -1610,7 +1600,7 @@ final class Replicators {
return new Replicator() {
@Override
protected Closeable applyTo(ChronicleMapBuilder builder,
protected Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap)
throws IOException {
return new UdpReplicator(map, entryExternalizable, replicationConfig,
@@ -78,12 +78,9 @@ final class SerializationBuilder<E> implements Cloneable {
this.eClass = eClass;
instancesAreMutable = instancesAreMutable(eClass);
ObjectFactory<E> factory = null;
if (role == Role.VALUE) {
factory = concreteClass(eClass) && marshallerUseFactory(eClass) ?
new NewInstanceObjectFactory<E>(eClass) :
(ObjectFactory<E>) NullObjectFactory.<E>of();
}
ObjectFactory<E> factory = concreteClass(eClass) && marshallerUseFactory(eClass) ?
new NewInstanceObjectFactory<E>(eClass) :
NullObjectFactory.<E>of();
if (concreteClass(eClass) && Byteable.class.isAssignableFrom(eClass)) {
agileMarshaller(ByteableMarshaller.of((Class) eClass), factory);
@@ -109,7 +106,7 @@ final class SerializationBuilder<E> implements Cloneable {
if (marshaller != null)
marshaller(marshaller, factory);
}
if (role == Role.VALUE && concreteClass(eClass) && marshallerUseFactory(eClass)) {
if (concreteClass(eClass) && marshallerUseFactory(eClass)) {
factory(factory);
}
}
@@ -390,12 +390,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
public synchronized V getUsing(K key, V value) {
public synchronized V getUsing(K key, V usingValue) {
throw new UnsupportedOperationException("getUsing is not supported for stateless clients");
}
public synchronized V acquireUsing(K key, V value) {
public synchronized V acquireUsing(K key, V usingValue) {
throw new UnsupportedOperationException("getUsing is not supported for stateless clients");
}
@@ -419,6 +419,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private void writeEntries(Map<? extends K, ? extends V> map) {
final int numberOfEntries = map.size();
int numberOfEntriesReadSoFar = 0;
@@ -558,7 +559,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
assert bytes.limit() == bytes.capacity();
}
@@ -602,22 +602,35 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
@NotNull
public synchronized Set<Map.Entry<K, V>> entrySet() {
long sizeLocation = writeEvent(ENTRY_SET);
final long sizeLocation = writeEvent(ENTRY_SET);
final long startTime = System.currentTimeMillis();
final long transactionId = nextUniqueTransaction(startTime);
final long timeoutTime = System.currentTimeMillis() + builder.timeoutMs();
// get the data back from the server
Bytes in = blockingFetch(sizeLocation);
Bytes in = blockingFetch0(sizeLocation, transactionId, startTime);
long size = in.readStopBit();
LOG.info("size="+size);
Set<Map.Entry<K, V>> result = new HashSet<Map.Entry<K, V>>();
final Set<Map.Entry<K, V>> result = new HashSet<Map.Entry<K, V>>();
for (int i = 0; i < size; i++) {
K k = keyValueSerializer.readKey(in);
V v = keyValueSerializer.readValue(in);
LOG.info("k="+k);
for (; ; ) {
LOG.info("v="+v);
result.add(new Entry(k, v));
boolean hasMoreEntries = in.readBoolean();
LOG.info("entrySet - hasMoreEntries=" + hasMoreEntries);
// number of entries in the chunk
long size = in.readInt();
for (int i = 0; i < size; i++) {
K k = keyValueSerializer.readKey(in);
V v = keyValueSerializer.readValue(in);
result.add(new Entry(k, v));
}
if (!hasMoreEntries)
break;
in = blockingFetchReadOnly(timeoutTime, transactionId);
}
return result;
@@ -691,7 +704,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private Bytes blockingFetch(long sizeLocation) {
try {
return blockingFetchThrowable(sizeLocation, this.builder.timeoutMs());
long startTime = System.currentTimeMillis();
return blockingFetchThrowable(sizeLocation, this.builder.timeoutMs(),
nextUniqueTransaction(startTime), startTime);
} catch (IOException e) {
close();
throw new IORuntimeException(e);
@@ -702,11 +717,25 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
private Bytes blockingFetchThrowable(long sizeLocation, long timeOutMs) throws IOException {
long startTime = System.currentTimeMillis();
private Bytes blockingFetch0(long sizeLocation, final long transactionId, long startTime) {
try {
return blockingFetchThrowable(sizeLocation, this.builder.timeoutMs(), transactionId, startTime);
} catch (IOException e) {
close();
throw new IORuntimeException(e);
} catch (Exception e) {
close();
throw e;
}
}
private Bytes blockingFetchThrowable(long sizeLocation, long timeOutMs, final long transactionId, final long startTime) throws IOException {
long timeoutTime = startTime + timeOutMs + 1000000;
long transactionId = nextUniqueTransaction(startTime);
for (; ; ) {
@@ -724,28 +753,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
// send out all the bytes
send(bytes, timeoutTime);
bytes.clear();
bytes.buffer().clear();
return blockingFetch(timeoutTime, transactionId);
// the number of bytes in the response
int size = receive(SIZE_OF_SIZE, timeoutTime).readInt();
if (bytes.capacity() < size)
bytes = new ByteBufferBytes(allocateDirect(size));
receive(size, timeoutTime);
boolean isException = bytes.readBoolean();
long inTransactionId = bytes.readLong();
if (inTransactionId != transactionId)
throw new IllegalStateException("the received transaction-id=" + inTransactionId +
", does not match the expected transaction-id=" + transactionId);
if (isException)
throw (RuntimeException) bytes.readObject();
return bytes;
} catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) {
checkTimeout(timeoutTime);
clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress());
@@ -753,6 +762,49 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
}
private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) {
try {
return blockingFetch(timeoutTime, transactionId);
} catch (IOException e) {
close();
throw new IORuntimeException(e);
} catch (Exception e) {
close();
throw e;
}
}
private Bytes blockingFetch(long timeoutTime, long transactionId) throws IOException {
bytes.clear();
bytes.buffer().clear();
// the number of bytes in the response
int size = receive(SIZE_OF_SIZE, timeoutTime).readInt();
int requiredSize = size + SIZE_OF_SIZE;
if (bytes.capacity() < requiredSize) {
bytes = new ByteBufferBytes(allocateDirect(requiredSize));
}
// block until we have received all the byte in this chunk
receive(size, timeoutTime);
boolean isException = bytes.readBoolean();
long inTransactionId = bytes.readLong();
if (inTransactionId != transactionId)
throw new IllegalStateException("the received transaction-id=" + inTransactionId +
", does not match the expected transaction-id=" + transactionId);
if (isException)
throw (RuntimeException) bytes.readObject();
return bytes;
}
private void write(long transactionId) {
bytes.writeLong(transactionId);
}
@@ -548,9 +548,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
if (completed)
attached.entryWriter.workCompleted();
// attached.hasRemoteHeartbeatInterval = false;
// attached.hasRemoteHeartbeatInterval = false;
} else if (attached.remoteModificationIterator != null)
} else if (attached.remoteModificationIterator != null)
attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key);
try {
@@ -1646,15 +1646,18 @@ class StatelessServerConnector<K, V> {
private Work entrySet(Bytes reader, Bytes writer) {
final long sizeLocation = reflectTransactionId(reader, writer);
final long transactionId = reader.readLong();
final Set<Map.Entry<K, V>> entries;
try {
entries = map.entrySet();
} catch (RuntimeException e) {
final long sizeLocation = reflectTransactionId(reader, writer);
return sendException(writer, sizeLocation, e);
}
writer.writeStopBit(entries.size());
final Iterator<Map.Entry<K, V>> iterator = entries.iterator();
@@ -1664,23 +1667,80 @@ class StatelessServerConnector<K, V> {
@Override
public boolean doWork(Bytes writer) {
final long sizeLocation = header(writer, transactionId);
int count = 0;
while (iterator.hasNext()) {
// we've filled up the buffer, so lets give another channel a chance to send
// some data, we don't know the max key size, we will use the entrySize instead
if (writer.remaining() <= maxEntrySizeBytes)
if (writer.remaining() <= maxEntrySizeBytes) {
writeHeader(writer, sizeLocation, count, true);
LOG.info("One more chunk !");
return false;
}
count++;
final Map.Entry<K, V> next = iterator.next();
writeKey(next.getKey(), writer);
writeValue(next.getValue(), writer);
}
writeSizeAndFlags(sizeLocation, false, writer);
writeHeader(writer, sizeLocation, count, false);
LOG.info("Last chunk !");
return true;
}
};
}
private void writeMoreData(long location, boolean hasMoreData, Bytes out) {
out.writeBoolean(location, hasMoreData);
}
private void entryCount(long location, int count, Bytes out) {
}
private long header(Bytes writer, final long transactionId) {
final long sizeLocation = writer.position();
writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // SIZE_OF_SIZE + is
// exception
writer.writeLong(transactionId);
// hasAnotherChunk
writer.skip(1);
// count
writer.skip(4);
return sizeLocation;
}
private void writeHeader(Bytes writer, long sizeLocation, int count, final boolean hasAnotherChunk) {
final long end = writer.position();
final int size = (int) (end - sizeLocation);
writer.position(sizeLocation);
// size in bytes
writer.writeInt(size);
// is exception
writer.writeBoolean(false);
//transaction id;
writer.skip(8);
writer.writeBoolean(hasAnotherChunk);
// count
writer.writeInt(count);
writer.position(end);
}
private Work putIfAbsent(Bytes reader, Bytes writer) {
final K key = readKey(reader);
final V v = readValue(reader);
@@ -1726,7 +1786,7 @@ class StatelessServerConnector<K, V> {
private long reflectTransactionId(Bytes reader, Bytes writer) {
final long sizeLocation = writer.position();
writer.skip(1 + AbstractChannelReplicator.SIZE_OF_SIZE); // event size + SIZE_OF_SIZE
writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // isException + SIZE_OF_SIZE
final long transactionId = reader.readLong();
writer.writeLong(transactionId);
return sizeLocation;
@@ -49,7 +49,7 @@ class UdpReplicator extends UdpChannelReplicator implements Replica.Modification
super(replicationConfig, serializedEntrySize, replica.identifier());
Replica.ModificationIterator modificationIterator = replica.acquireModificationIterator(
ChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID, this);
AbstractChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID, this);
setReader(new UdpSocketChannelEntryReader(serializedEntrySize, entryExternalizable));
@@ -76,6 +76,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
final MetaProvider<V, VW, MVW> metaValueWriterProvider;
final ObjectFactory<V> valueFactory;
final DefaultValueProvider<K, V> defaultValueProvider;
final PrepareValueBytesAsWriter<K> prepareValueBytesAsWriter;
final int metaDataBytes;
// private final int replicas;
@@ -101,7 +102,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
private final HashSplitting hashSplitting;
public VanillaChronicleMap(ChronicleMapBuilder<K, V> builder) throws IOException {
public VanillaChronicleMap(AbstractChronicleMapBuilder<K, V, ?> builder) throws IOException {
SerializationBuilder<K> keyBuilder = builder.keyBuilder;
kClass = keyBuilder.eClass;
@@ -123,6 +124,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
metaValueWriterProvider = (MetaProvider) valueBuilder.metaInteropProvider();
valueFactory = valueBuilder.factory();
defaultValueProvider = builder.defaultValueProvider();
prepareValueBytesAsWriter = builder.prepareValueBytesAsWriter();
lockTimeOutNS = builder.lockTimeOut(TimeUnit.NANOSECONDS);
@@ -356,13 +358,13 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
@Override
public V getUsing(K key, V value) {
return lookupUsing(key, value, false);
public V getUsing(K key, V usingValue) {
return lookupUsing(key, usingValue, false);
}
@Override
public V acquireUsing(K key, V value) {
return lookupUsing(key, value, true);
public V acquireUsing(K key, V usingValue) {
return lookupUsing(key, usingValue, true);
}
V lookupUsing(K key, V value, boolean create) {
@@ -713,18 +715,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return (int) (sizeInBytes / entrySize) + 1;
}
/**
* Used to acquire an object of type V from the Segment. {@code usingValue} is reused to read the
* value if key is present in this Segment, if key is absent in this Segment: <ol><li>If {@code
* create == false}, just {@code null} is returned (except when event listener provides a value "on
* get missing" - then it is put into this Segment for the key).</li> <li>If {@code create ==
* true}, {@code usingValue} or a newly created instance of value class, if {@code usingValue ==
* null}, is put into this Segment for the key.</li></ol>
*
* @param hash2 a hash code related to the {@code keyBytes}
* @return the value which is finally associated with the given key in this Segment after execution of
* this method, or {@code null}.
*/
V acquire(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue,
long hash2, boolean create) {
if (create)
@@ -739,22 +729,23 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
if (offset >= 0L) {
return onKeyPresentOnAcquire(copies, key, usingValue, offset, entry);
} else {
boolean usingValuePassed = usingValue != null;
usingValue = tryObtainUsingValueOnAcquire(key, usingValue, create);
if (usingValue != null) {
// If `create` is false, this method was called from get() or getUsing()
// and non-null `usingValue` was returned by notifyMissed() method.
// This "missed" default value is considered as genuine value
// rather than "using" container to fill up, even if it implements Byteable.
offset = putEntry(copies, metaKeyInterop, keyInterop, key, keySize,
usingValue, create);
incrementSize();
if (usingValuePassed || !create)
notifyPut(offset, true, key, usingValue, posFromOffset(offset));
return usingValue;
} else {
if (!create)
return null;
if (defaultValueProvider != null) {
V defaultValue = defaultValueProvider.get(key);
copies = valueWriterProvider.getCopies(copies);
VW valueWriter = valueWriterProvider.get(copies, originalValueWriter);
copies = metaValueWriterProvider.getCopies(copies);
MetaBytesWriter<V, VW> metaValueWriter = metaValueWriterProvider.get(
copies, originalMetaValueWriter, valueWriter, defaultValue);
offset = putEntry(metaKeyInterop, keyInterop, key, keySize,
metaValueWriter, valueWriter, defaultValue);
} else {
offset = putEntry(metaKeyInterop, keyInterop, key, keySize,
prepareValueBytesAsWriter, null, key);
}
return onKeyAbsentOnAcquire(copies, key, keySize, usingValue, offset);
}
} finally {
if (create)
@@ -782,36 +773,26 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
V onKeyPresentOnAcquire(ThreadLocalCopies copies, K key, V usingValue, long offset,
MultiStoreBytes entry) {
skipReplicationBytes(entry);
V v = readValue(copies, entry, usingValue);
notifyGet(offset, key, v);
return v;
}
V tryObtainUsingValueOnAcquire(K key, V usingValue, boolean create) {
if (create) {
if (usingValue != null) {
return usingValue;
} else {
try {
usingValue = valueFactory.create();
if (usingValue == null) {
throw new IllegalStateException("acquireUsing() method requires" +
"valueFactory.create() result to be non-null. " +
"By default it is so when value class is" +
"a Byteable/BytesMarshallable/Externalizable subclass." +
"Note that acquireUsing() anyway makes very little sense " +
"when value class is not a Byteable subclass.");
}
return usingValue;
} catch (Exception e) {
throw new IllegalStateException(e);
}
}
} else {
if (usingValue instanceof Byteable)
((Byteable) usingValue).bytes(null, 0L);
return defaultValueProvider.get(key, usingValue);
}
V onKeyAbsentOnAcquire(ThreadLocalCopies copies, K key, long keySize, V usingValue,
long offset) {
incrementSize();
MultiStoreBytes entry = entry(offset);
entry.skip(keySizeMarshaller.sizeEncodingSize(keySize) + keySize);
skipReplicationBytes(entry);
V v = readValue(copies, entry, usingValue);
notifyPut(offset, true, key, v, posFromOffset(offset));
notifyGet(offset, key, v);
return v;
}
void skipReplicationBytes(Bytes bytes) {
// don't skip
}
V put(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V value,
@@ -835,8 +816,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
// key is not found
long offset = putEntry(copies, metaKeyInterop, keyInterop, key, keySize, value,
false);
long offset = putEntry(copies, metaKeyInterop, keyInterop, key, keySize, value);
incrementSize();
notifyPut(offset, true, key, value, posFromOffset(offset));
return null;
@@ -861,37 +841,22 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return prevValue;
}
/**
* Puts entry. If {@code value} implements {@link Byteable} interface and {@code usingValue} is {@code
* true}, the value is backed with the bytes of this entry.
*
* @param value the value to put
* @param usingValue {@code true} if the value should be backed with the bytes of the entry, if it
* implements {@link Byteable} interface, {@code false} if it should put itself
* @return offset of the written entry in the Segment bytes
*/
private long putEntry(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key,
long keySize, V value, boolean usingValue) {
long keySize, V value) {
copies = valueWriterProvider.getCopies(copies);
VW valueWriter = valueWriterProvider.get(copies, originalValueWriter);
copies = metaValueWriterProvider.getCopies(copies);
MetaBytesWriter<V, VW> metaValueWriter = metaValueWriterProvider.get(
copies, originalMetaValueWriter, valueWriter, value);
// "if-else polymorphism" is not very beautiful, but allows to
// reuse the rest code of this method and doesn't hurt performance.
boolean byteableValue = usingValue && value instanceof Byteable;
long valueSize;
MetaBytesWriter<V, VW> metaValueWriter = null;
VW valueWriter = null;
Byteable valueAsByteable = null;
if (!byteableValue) {
copies = valueWriterProvider.getCopies(copies);
valueWriter = valueWriterProvider.get(copies, originalValueWriter);
copies = metaValueWriterProvider.getCopies(copies);
metaValueWriter = metaValueWriterProvider.get(
copies, originalMetaValueWriter, valueWriter, value);
valueSize = metaValueWriter.size(valueWriter, value);
} else {
valueAsByteable = (Byteable) value;
valueSize = valueAsByteable.maxSize();
}
return putEntry(metaKeyInterop, keyInterop, key, keySize,
metaValueWriter, valueWriter, value);
}
private <E, EW> long putEntry(MKI metaKeyInterop, KI keyInterop, K key, long keySize,
MetaBytesWriter<E, EW> metaElemWriter, EW elemWriter,
E elem) {
long valueSize = metaElemWriter.size(elemWriter, elem);
long entrySize = entrySize(keySize, valueSize);
long pos = alloc(inBlocks(entrySize));
long offset = offsetFromPos(pos);
@@ -900,27 +865,13 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
keySizeMarshaller.writeSize(entry, keySize);
metaKeyInterop.write(keyInterop, entry, key);
writeValueOnPutEntry(valueSize, metaValueWriter, valueWriter, value, valueAsByteable,
entry);
hashLookup.putAfterFailedSearch(pos);
return offset;
}
void writeValueOnPutEntry(long valueSize, MetaBytesWriter<V, VW> metaValueWriter,
VW valueWriter, V value, Byteable valueAsByteable,
NativeBytes entry) {
valueSizeMarshaller.writeSize(entry, valueSize);
alignment.alignPositionAddr(entry);
metaElemWriter.write(elemWriter, entry, elem);
if (metaValueWriter != null) {
metaValueWriter.write(valueWriter, entry, value);
} else {
assert valueAsByteable != null;
long valueOffset = entry.positionAddr() - bytes.address();
bytes.zeroOut(valueOffset, valueOffset + valueSize);
valueAsByteable.bytes(bytes, valueOffset);
}
hashLookup.putAfterFailedSearch(pos);
return offset;
}
void clearMetaData(long offset) {
@@ -0,0 +1,34 @@
/*
* 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.io.Bytes;
final class ZeroOutValueBytes<K> implements PrepareValueBytes<K> {
private static final long serialVersionUID = 0L;
private final long valueSize;
ZeroOutValueBytes(long valueSize) {
this.valueSize = valueSize;
}
@Override
public void prepare(Bytes bytes, K key) {
long pos = bytes.position();
bytes.zeroOut(pos, pos + valueSize);
}
}
@@ -25,6 +25,7 @@ import net.openhft.chronicle.common.TimeProvider;
import net.openhft.chronicle.map.*;
import net.openhft.lang.io.serialization.BytesMarshaller;
import net.openhft.lang.io.serialization.BytesMarshallerFactory;
import net.openhft.lang.io.serialization.ObjectFactory;
import net.openhft.lang.io.serialization.ObjectSerializer;
import net.openhft.lang.io.serialization.impl.NullObjectFactory;
import org.jetbrains.annotations.NotNull;
@@ -52,8 +53,7 @@ public class ChronicleSetBuilder<E>
ChronicleSetBuilder(Class<E> keyClass) {
chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, DummyValue.class)
.entryAndValueAlignment(Alignment.NO_ALIGNMENT)
.valueMarshallerAndFactory(DummyValueMarshaller.INSTANCE,
NullObjectFactory.<DummyValue>of());
.valueMarshaller(DummyValueMarshaller.INSTANCE);
}
public static <K> ChronicleSetBuilder<K> of(Class<K> keyClass) {
@@ -244,6 +244,23 @@ public class ChronicleSetBuilder<E>
return this;
}
/**
* {@inheritDoc}
*
* <p>Actually this is just a convenience method supporting key marshaller configurations, made
* initially during {@link #of(Class)} call. Because if you {@linkplain
* #keyMarshaller(BytesMarshaller) configure} own custom key marshaller, this method doesn't
* take any effect on the maps constructed by this builder.
*
* @see #of(Class)
*/
@Override
public ChronicleSetBuilder<E> keyDeserializationFactory(
@NotNull ObjectFactory<E> keyDeserializationFactory) {
chronicleMapBuilder.keyDeserializationFactory(keyDeserializationFactory);
return this;
}
@Override
public ChronicleSetBuilder<E> immutableKeys() {
chronicleMapBuilder.immutableKeys();
@@ -0,0 +1,21 @@
/*
* 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.
*/
/**
* Contains {@link net.openhft.chronicle.set.ChronicleSet} interface and
* {@link net.openhft.chronicle.set.ChronicleSetBuilder} class.
*/
package net.openhft.chronicle.set;