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

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Peter Lawrey
2014-12-04 09:58:27 +02:00
parent fbafa8fc67 8d28d0194d
commit c83c065c77
31 changed files with 1220 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pom.xml
+4 -4
View File
@@ -30,7 +30,7 @@
<modelVersion>4.0.0</modelVersion>
<artifactId>chronicle-map</artifactId>
<version>2.0.11b-SNAPSHOT</version>
<version>2.0.12b-SNAPSHOT</version>
<name>OpenHFT/Chronicle-Map</name>
<description>Chronicle-Map</description>
@@ -48,13 +48,13 @@
<dependency>
<groupId>net.openhft</groupId>
<artifactId>lang</artifactId>
<version>6.4.15-SNAPSHOT</version>
<version>6.4.16-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>net.openhft</groupId>
<artifactId>affinity</artifactId>
<version>2.1.3</version>
<version>2.1.4</version>
</dependency>
<dependency>
@@ -167,7 +167,7 @@
<artifactId>maven-javadoc-plugin</artifactId>
<configuration>
<links>
<link>http://openhft.github.io/Chronicle-Map/apidocs/</link>
<link>http://openhft.github.io/Java-Lang/apidocs/</link>
</links>
<excludePackageNames>net.openhft.chronicle.hash.serialization.internal
</excludePackageNames>
src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
+3 -3
View File
@@ -46,7 +46,7 @@ import java.util.concurrent.TimeUnit;
/**
* This interface defines the meaning of configurations, common to {@link
* net.openhft.chronicle.map.OnHeapUpdatableChronicleMapBuilder} and {@link ChronicleSetBuilder}, i.
* net.openhft.chronicle.map.ChronicleMapBuilder} and {@link ChronicleSetBuilder}, i.
* e. <i>Chronicle hash container</i> configurations.
*
* <p>{@code ChronicleHashBuilder} is mutable. Configuration methods mutate the builder and return
@@ -131,7 +131,7 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
*
* <p>There are three major patterns of this configuration usage: <ol> <li>Key (and value, in
* {@code ChronicleMap} case) sizes are constant. Configure them via {@link
* #constantKeySizeBySample(Object)} and {@link net.openhft.chronicle.map.OnHeapUpdatableChronicleMapBuilder#constantValueSizeBySample(Object)}
* #constantKeySizeBySample(Object)} and {@link net.openhft.chronicle.map.ChronicleMapBuilder#constantValueSizeBySample(Object)}
* methods, and you will experience no memory waste at all.</li> <li>Key (and/or value size, in
* {@code ChronicleMap} case) varies moderately. Specify them using corresponding methods, or
* specify entry size directly by calling this method, by sizes somewhere between average and
@@ -184,7 +184,7 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* memory unit).</li> <li>If key (and/or value, in {@code ChronicleMap} case) size varies
* moderately, you should pass to this method the maximum number of entries + 5-25%, depending
* on your data properties and configured {@linkplain #keySize(int) key}/{@linkplain
* net.openhft.chronicle.map.OnHeapUpdatableChronicleMapBuilder#valueSize(int)
* net.openhft.chronicle.map.ChronicleMapBuilder#valueSize(int)
* value}/{@linkplain #entrySize(int) entry} sizes.</li> <li>If your data size varies in a wide
* range, pass the maximum number of entries multiplied by average data size and divided by the
* configured "entry size" (i. e. chunk size). See an example in the documentation to {@link
src/main/java/net/openhft/chronicle/hash/FindByName.java
+3 -1
View File
@@ -10,10 +10,12 @@ public interface FindByName {
/**
* @param name the name of the map or set
* @param <T>
* @param <T> the type returned
* @return a chronicle map or set
* @throws IllegalArgumentException if a map with this name can not be found
* @throws IOException if it not possible to create the map or set
* @throws TimeoutException if the call times out
* @throws InterruptedException if interrupted by another thread
*/
<T extends ChronicleHash> T from(String name) throws IllegalArgumentException,
IOException, TimeoutException, InterruptedException;
src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java
+7 -3
View File
@@ -135,7 +135,9 @@ public final class TcpTransportAndNetworkConfig implements Serializable {
}
/**
* @param name the name used here is reflected in the Tcp Replicator and is typically used when debugging
* @param name the name used here is reflected in the Tcp Replicator and is typically used when
* debugging
* @return self
*/
public TcpTransportAndNetworkConfig name(String name) {
this.name = name;
@@ -164,11 +166,13 @@ public final class TcpTransportAndNetworkConfig implements Serializable {
TcpTransportAndNetworkConfig that = (TcpTransportAndNetworkConfig) o;
if (autoReconnectedUponDroppedConnection != that.autoReconnectedUponDroppedConnection) return false;
if (autoReconnectedUponDroppedConnection != that.autoReconnectedUponDroppedConnection)
return false;
if (heartBeatInterval != that.heartBeatInterval) return false;
if (packetSize != that.packetSize) return false;
if (serverPort != that.serverPort) return false;
if (endpoints != null ? !endpoints.equals(that.endpoints) : that.endpoints != null) return false;
if (endpoints != null ? !endpoints.equals(that.endpoints) : that.endpoints != null)
return false;
if (heartBeatIntervalUnit != that.heartBeatIntervalUnit) return false;
if (throttlingConfig != null ? !throttlingConfig.equals(that.throttlingConfig) : that.throttlingConfig != null)
return false;
src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java
+8 -1
View File
@@ -103,7 +103,14 @@ public class ByteableMarshaller<E extends Byteable>
@Override
public void write(Bytes bytes, E e) {
bytes.write(e.bytes(), e.offset(), size);
Bytes eBytes = e.bytes();
if (eBytes != null) {
bytes.write(eBytes, e.offset(), size);
} else {
throw new NullPointerException("You are trying to write a byteable object of " +
e.getClass() + ", " +
"which bytes are not assigned. I. e. most likely the object is uninitialized.");
}
}
@Override
src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+1
View File
@@ -53,6 +53,7 @@ abstract class AbstractChannelReplicator implements Closeable {
public static final int SIZE_OF_SIZE = 4;
public static final int SIZE_OF_TRANSACTION_ID = 8;
public static final int SIZE_OF_IS_EXCEPTION = 1;
private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class);
src/main/java/net/openhft/chronicle/map/BytesMapEventListener.java
+1 -3
View File
@@ -18,8 +18,6 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.serialization.SizeMarshaller;
import net.openhft.lang.io.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Serializable;
@@ -29,7 +27,7 @@ import java.io.Serializable;
* {@linkplain Bytes#limit() limit}).</li>
* <li>Methods must not update entry bytes, except in metadata area</li>
* <li>From {@code metaDataPos} offset metadata area starts in the given {@code entry}, listener
* should be itself aware of it's length (see {@link ChronicleMapBuilder#metaDataBytes()}).</li>
* should be itself aware of it's length (see {@link ChronicleMapBuilder#metaDataBytes(int)}).</li>
* <li>From {@code keyPos} offset key area in the given {@code entry}, serialized key size
* (using {@linkplain ChronicleMapBuilder#keySizeMarshaller(SizeMarshaller)}), directly followed
* by the serialized key itself.</li>
src/main/java/net/openhft/chronicle/map/ChronicleMap.java
+13 -8
View File
@@ -220,6 +220,7 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
* Calling this method may or may not block until the operating is complete depending on the
* implementation.
*
* @param key the key whose associated value is to be returned in the future
* @return a future containing the value to which the specified key is mapped after this method
* call, or {@code null} if no value is mapped
* @see java.util.Map#get(java.lang.Object)
@@ -236,10 +237,14 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
* Calling this method may or may not block until the operating is complete depending on the
* implementation.
*
* @return a future containing the previous value associated with <tt>key</tt>, or <tt>null</tt>
* if there was no mapping for <tt>key</tt>. (A <tt>null</tt> return can also indicate that the
* map previously associated <tt>null</tt> with <tt>key</tt>, if the implementation supports
* <tt>null</tt> values.)
* @param key key with which the specified value is to be associated
* @param value value to be associated with the specified key * @return a future containing the
* previous value associated with <tt>key</tt>, or <tt>null</tt> if there was no
* mapping for <tt>key</tt>. (A <tt>null</tt> return can also indicate that the map
* previously associated <tt>null</tt> with <tt>key</tt>, if the implementation
* supports <tt>null</tt> values.)
* @return a future containing either, the previous value associated with <tt>key</tt>, or
* <tt>null</tt> if there was no mapping for <tt>key</tt>.
* @see java.util.Map#put(java.lang.Object, java.lang.Object)
* @see java.util.concurrent.Future
*/
@@ -279,8 +284,8 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
* differ. This method is not performant and as such we recommend it is not used in performance
* sensitive code.
*
* MAPS CONTAINING KEYS/VALUES GENERATED WITH OFF-HEAP INTERFACES ARE NOT CURRENTLY
* SUPPORTED, CURRENTLY ONLY KEYS/VALUES THAT IMPLEMENT SERIALIZABLE ARE SUPPORTED
* MAPS CONTAINING KEYS/VALUES GENERATED WITH OFF-HEAP INTERFACES ARE NOT CURRENTLY SUPPORTED,
* CURRENTLY ONLY KEYS/VALUES THAT IMPLEMENT SERIALIZABLE ARE SUPPORTED
*
* @param toFile the file to store all the entries to, the entries will be stored in JSON
* format
@@ -296,8 +301,8 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
* entries are overwritten. A write lock is only held while each individual entry is inserted
* into the map, not over all the entries in the {@link java.io.File}
*
* MAPS CONTAINING KEYS/VALUES GENERATED WITH OFF-HEAP INTERFACES ARE NOT CURRENTLY
* SUPPORTED, CURRENTLY ONLY KEYS/VALUES THAT IMPLEMENT SERIALIZABLE ARE SUPPORTED
* MAPS CONTAINING KEYS/VALUES GENERATED WITH OFF-HEAP INTERFACES ARE NOT CURRENTLY SUPPORTED,
* CURRENTLY ONLY KEYS/VALUES THAT IMPLEMENT SERIALIZABLE ARE SUPPORTED
*
* @param fromFile the file containing entries ( in JSON format ) which will be deserialized and
* {@link java.util.Map#put(java.lang.Object, java.lang.Object)} into the map
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java
+75 -67
View File
@@ -30,25 +30,25 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
* {@inheritDoc} 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, and entry
* alignment is set to {@link Alignment#OF_4_BYTES}, the actual entry size will be 32 (30 aligned to 4
* bytes).
* #entryAndValueAlignment(Alignment) entry alignment}). I. e. if you set entry size to 30, and
* entry alignment is set to {@link Alignment#OF_4_BYTES}, the actual entry size will be 32 (30
* aligned to 4 bytes).
*
* @see #entryAndValueAlignment(Alignment) // * @see #entries(long)
*/
ChronicleMapBuilderI<K, V> entrySize(int entrySize);
/**
* 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.
* 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>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>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} for Byteable values.
*
@@ -58,11 +58,11 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
ChronicleMapBuilderI<K, V> entryAndValueAlignment(Alignment alignment);
/**
* {@inheritDoc} Also, it overrides any previous {@link #prepareValueBytesOnAcquire} configuration to this
* {@code ChronicleMapBuilder}.
* {@inheritDoc} Also, it overrides any previous {@link #prepareValueBytesOnAcquire}
* configuration to this {@code ChronicleMapBuilder}.
*
* <p>By default, the default value is not specified, default {@linkplain #prepareValueBytesOnAcquire
* prepare value bytes routine} is specified instead.
* <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)
@@ -70,15 +70,16 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
ChronicleMapBuilderI<K, V> defaultValue(V defaultValue);
/**
* 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. See {@link PrepareValueBytes} for more information.
* 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. See {@link PrepareValueBytes} for more
* information.
*
* <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
* net.openhft.lang.model.Byteable} value to return from {@code acquireUsing()}
* call
* @param prepareValueBytes what to do with the value bytes before assigning them into the
* {@link net.openhft.lang.model.Byteable} value to return from {@code
* acquireUsing()} call
* @return this builder back
* @see PrepareValueBytes
* @see #defaultValue(Object)
@@ -106,16 +107,17 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
ChronicleMapBuilderI<K, V> constantKeySizeBySample(K sampleKey);
/**
* Configures the optimal number of bytes, taken by serialized form of values, put into maps, created by
* this builder. If value size is always the same, call {@link #constantValueSizeBySample(Object)} method
* instead of this one.
* Configures the optimal number of bytes, taken by serialized form of values, put into maps,
* created by this builder. If value size is always the same, call {@link
* #constantValueSizeBySample(Object)} method instead of this one.
*
* <p>If value is a boxed primitive type, i. e. if value size is known statically, it is automatically
* accounted and shouldn't be specified by user.
* <p>If value is a boxed primitive type, i. e. if value size is known statically, it is
* automatically accounted and shouldn't be specified by user.
*
* <p>If value size varies moderately, specify the size higher than average, but lower than the maximum
* possible, to minimize average memory overuse. If value size varies in a wide range, it's better to use
* {@linkplain #entrySize(int) entry size} in "chunk" mode and configure it directly.
* <p>If value size varies moderately, specify the size higher than average, but lower than the
* maximum possible, to minimize average memory overuse. If value size varies in a wide range,
* it's better to use {@linkplain #entrySize(int) entry size} in "chunk" mode and configure it
* directly.
*
* @param valueSize number of bytes, taken by serialized form of values
* @return this {@code ChronicleMapOnHeapUpdatableBuilder} back
@@ -127,15 +129,16 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
ChronicleMapBuilderI<K, V> valueSize(int valueSize);
/**
* Configures the constant number of bytes, taken by serialized form of values, put into maps, created by
* this builder. This is done by providing the {@code sampleValue}, all values should take the same number
* of bytes in serialized form, as this sample object.
* Configures the constant number of bytes, taken by serialized form of values, put into maps,
* created by this builder. This is done by providing the {@code sampleValue}, all values should
* take the same number of bytes in serialized form, as this sample object.
*
* <p>If values are of boxed primitive type or {@link net.openhft.lang.model.Byteable} subclass, i. e. if
* value size is known statically, it is automatically accounted and this method shouldn't be called.
* <p>If values are of boxed primitive type or {@link net.openhft.lang.model.Byteable} subclass,
* i. e. if value size is known statically, it is automatically accounted and this method
* shouldn't be called.
*
* <p>If value size varies, method {@link #valueSize(int)} or {@link #entrySize(int)} should be called
* instead of this one.
* <p>If value size varies, method {@link #valueSize(int)} or {@link #entrySize(int)} should be
* called instead of this one.
*
* @param sampleValue the sample value
* @return this builder back
@@ -164,8 +167,8 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
/**
* {@inheritDoc}
*
* @throws IllegalStateException if custom value marshaller is specified or value class is not either
* {@code BytesMarshallable} or {@code Externalizable}
* @throws IllegalStateException if custom value marshaller is specified or value class is not
* either {@code BytesMarshallable} or {@code Externalizable}
*/
ChronicleMapBuilderI<K, V> valueDeserializationFactory(
@@ -174,8 +177,8 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
/**
* {@inheritDoc}
*
* <p>By default, default value provider is not specified, {@link #defaultValue(Object) default value} is
* specified instead.
* <p>By default, default value provider is not specified, {@link #defaultValue(Object) default
* value} is specified instead.
*
* @see #defaultValue(Object)
*/
@@ -184,8 +187,8 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
@NotNull DefaultValueProvider<K, V> defaultValueProvider);
/**
* Configures the {@code BytesMarshaller} used to serialize/deserialize values to/from off-heap memory in
* maps, created by this builder. See <a href="https://github.com/OpenHFT/Chronicle-Map#serialization">the
* Configures the {@code BytesMarshaller} used to serialize/deserialize values to/from off-heap
* memory in maps, created by this builder. See <a href="https://github.com/OpenHFT/Chronicle-Map#serialization">the
* section about serialization in ChronicleMap manual</a> for more information.
*
* @param valueMarshaller the marshaller used to serialize values
@@ -199,20 +202,21 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
@NotNull BytesMarshaller<V> valueMarshaller);
/**
* Configures the marshallers, used to serialize/deserialize values to/from off-heap memory in maps,
* created by this builder. See <a href="https://github.com/OpenHFT/Chronicle-Map#serialization">the
* Configures the marshallers, used to serialize/deserialize values to/from off-heap memory in
* maps, created by this builder. See <a href="https://github.com/OpenHFT/Chronicle-Map#serialization">the
* section about serialization in ChronicleMap manual</a> for more information.
*
* <p>Configuring marshalling this way results to a little bit more compact in-memory layout of the map,
* comparing to a single interface configuration: {@link #valueMarshaller(BytesMarshaller)}.
* <p>Configuring marshalling this way results to a little bit more compact in-memory layout of
* the map, comparing to a single interface configuration: {@link #valueMarshaller(BytesMarshaller)}.
*
* <p>Passing {@link net.openhft.chronicle.hash.serialization.BytesInterop} instead of plain {@link
* net.openhft.chronicle.hash.serialization.BytesWriter} is, of cause, possible, but currently pointless
* for values.
* <p>Passing {@link net.openhft.chronicle.hash.serialization.BytesInterop} instead of plain
* {@link net.openhft.chronicle.hash.serialization.BytesWriter} is, of cause, possible, but
* currently pointless for values.
*
* @param valueWriter the new value object &rarr; {@link net.openhft.lang.io.Bytes} writer (interop)
* @param valueWriter the new value object &rarr; {@link net.openhft.lang.io.Bytes} writer
* (interop) strategy
* @param valueReader the new {@link net.openhft.lang.io.Bytes} &rarr; value object reader
* strategy
* @param valueReader the new {@link net.openhft.lang.io.Bytes} &rarr; value object reader strategy
* @return this builder back
* @see #valueMarshaller(BytesMarshaller)
*/
@@ -220,19 +224,21 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
@NotNull BytesWriter<V> valueWriter, @NotNull BytesReader<V> valueReader);
/**
* Configures the marshaller used to serialize actual value sizes to off-heap memory in maps, created by
* this builder.
* Configures the marshaller used to serialize actual value sizes to off-heap memory in maps,
* created by this builder.
*
* <p>Default value size marshaller is so-called {@linkplain net.openhft.chronicle.hash.serialization.SizeMarshallers#stopBit()
* stop bit encoding marshalling}. If {@linkplain #constantValueSizeBySample(Object) constant value size}
* is configured, or defaulted if the value type is always constant and {@code ChronicleHashBuilder}
* implementation knows about it, this configuration takes no effect, because a special {@link
* net.openhft.chronicle.hash.serialization.SizeMarshaller} implementation, which doesn't actually do any
* marshalling, and just returns the known constant size on {@link net.openhft.chronicle.hash.serialization.SizeMarshaller#readSize(
*net.openhft.lang.io.Bytes)} calls, is used instead of any {@code SizeMarshaller} configured using this
* method.
* stop bit encoding marshalling}. If {@linkplain #constantValueSizeBySample(Object) constant
* value size} is configured, or defaulted if the value type is always constant and {@code
* ChronicleHashBuilder} implementation knows about it, this configuration takes no effect,
* because a special {@link net.openhft.chronicle.hash.serialization.SizeMarshaller}
* implementation, which doesn't actually do any marshalling, and just returns the known
* constant size on {@link net.openhft.chronicle.hash.serialization.SizeMarshaller#readSize(
*net.openhft.lang.io.Bytes)} calls, is used instead of any {@code SizeMarshaller} configured
* using this method.
*
* @param valueSizeMarshaller the new marshaller, used to serialize actual value sizes to off-heap memory
* @param valueSizeMarshaller the new marshaller, used to serialize actual value sizes to
* off-heap memory
* @return this builder back
*/
ChronicleMapBuilderI<K, V> valueSizeMarshaller(
@@ -295,13 +301,15 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
ChronicleMapBuilderI<K, V> name(String name);
/**
* @param bootstapOnlyLocalEntries if set to true - when a new node joins a TCP replication grid, the new
* node will be populated with data, only for the nodes that created that
* data. Otherwise, all the nodes will publish all the data they have (
* potentially swamping the new node with duplicates ) however this does
* guarantee that all the data is replicated over to the new node, and is
* useful especially in the case that the originating node is not
* @param bootstapOnlyLocalEntries if set to true - when a new node joins a TCP replication
* grid, the new node will be populated with data, only for the
* nodes that created that data. Otherwise, all the nodes will
* publish all the data they have ( potentially swamping the new
* node with duplicates ) however this does guarantee that all
* the data is replicated over to the new node, and is useful
* especially in the case that the originating node is not
* currently running.
* @return self
*/
ChronicleMapBuilderI<K, V> bootstrapOnlyLocalEntries(boolean bootstapOnlyLocalEntries);
src/main/java/net/openhft/chronicle/map/MapEventListener.java
+31 -27
View File
@@ -26,22 +26,22 @@ import org.slf4j.LoggerFactory;
import java.io.Serializable;
/**
* Contains methods which are called when {@link ChronicleMap} key events occur. Typical use cases: <ul>
* <li>Map data backup / replication.</li> <li>Logging, monitoring, debugging.</li> </ul>
* Contains methods which are called when {@link ChronicleMap} key events occur. Typical use cases:
* <ul> <li>Map data backup / replication.</li> <li>Logging, monitoring, debugging.</li> </ul>
*
* <p>This is an adapter class - all methods have default implementations as no-ops. Extend this class and
* override only methods corresponding the events you are interested in.
* <p>This is an adapter class - all methods have default implementations as no-ops. Extend this
* class and override only methods corresponding the events you are interested in.
*
* <p>To configure {@code MapEventListener} for {@code ChronicleMap}, use {@link
* AbstractChronicleMapBuilder#eventListener(MapEventListener)} method.
*
* <p>See {@link #logging(String)} implementation.
*
* <p>All these calls are synchronous while a {@code ChronicleMap} lock is held so make them as quick as
* possible.
* <p>All these calls are synchronous while a {@code ChronicleMap} lock is held so make them as
* quick as possible.
*
* <p>The {@code entry} passed to the methods is {@code Bytes} instance positioned at meta data area. See
* {@link ChronicleHashBuilder#metaDataBytes(int)} for more information.
* <p>The {@code entry} passed to the methods is {@code Bytes} instance positioned at meta data
* area. See {@link ChronicleHashBuilder#metaDataBytes(int)} for more information.
*
* @param <K> key type of the maps, trackable by this event listener
* @param <V> value type of the maps, trackable by this event listener
@@ -51,15 +51,17 @@ public abstract class MapEventListener<K, V> implements Serializable {
private static final long serialVersionUID = 0L;
/**
* Returns the map event listener, which logs strings like
* "{@linkplain ChronicleMap#file() map file} opType key value", where opType is either "get",
* "put" or "remove", to the logger provided by SLF4J.
* Returns the map event listener, which logs strings like "{@linkplain ChronicleMap#file() map
* file} opType key value", where opType is either "get", "put" or "remove", to the logger
* provided by SLF4J.
*
* @param <K> the map key type
* @param <V> the map value type
* @param <K> the map key type
* @param <V> the map value type
* @param prefix logging prefix
* @return the logging event listener
*/
public static <K, V> MapEventListener<K, V> logging(String prefix) {
//noinspection unchecked
return new LoggingMapEventListener(prefix);
}
@@ -90,40 +92,42 @@ public abstract class MapEventListener<K, V> implements Serializable {
}
/**
* 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.
* 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 key the key looked up
* @param foundValue the value found for the key
* @param key the key looked up
* @param foundValue the value found for the key
*/
public void onGetFound(K key, V foundValue) {
// do nothing
}
/**
* This method is called 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
* This method is called 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
* 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.
* 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.
*
* <p>This method is called when put is already happened.
*
* @param key the key the given value is put for
* @param newValue the value which is now associated with the given key
* @param replacedValue the value which was replaced by {@code newValue}, {@code null}
* if the key was absent in the map before current {@code ChronicleMap}
* @param replacedValue the value which was replaced by {@code newValue}, {@code null} if the
* key was absent in the map before current {@code ChronicleMap}
*/
public void onPut(K key, V newValue, @Nullable V replacedValue) {
// do nothing
}
/**
* This is called when an entry is removed. Misses, i. e. when {@code map.remove(key)}
* is called, but key is already absent in the map, are not notified.
* This is called when an entry is removed. Misses, i. e. when {@code map.remove(key)} is
* called, but key is already absent in the map, are not notified.
*
* @param key the key removed from the map
* @param value the value which was associated with the given key
* @param key the key removed from the map
* @param value the value which was associated with the given key
*/
public void onRemove(K key, V value) {
// do nothing
src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java
+7 -7
View File
@@ -36,7 +36,7 @@ import java.io.Serializable;
* non-{@link Byteable} values, which don't point to off-heap memory directly, including primitives ({@link
* Long}, {@link Double}, etc.), {@link String}s and {@link CharSequence}s, values implementing {@link
* BytesMarshallable}, {@link Externalizable} or {@link Serializable} interface, or any other values for which
* {@linkplain #valueMarshaller(BytesMarshaller) custom marshaller} is provided. <p/> <p>Use static {@link
* {@linkplain #valueMarshaller(BytesMarshaller) custom marshaller} is provided. Use static {@link
* #of(Class, Class) of(Key.class, Value.class)} method to obtain a {@code ChronicleMapOnHeapUpdatableBuilder}
* instance.
*
@@ -103,8 +103,8 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
/**
* Configures the optimal number of bytes, taken by serialized form of values, put into maps, created by
* this builder. If value size is always the same, call {@link #constantValueSizeBySample(Object)} method
* instead of this one. <p/> <p>If value is a boxed primitive type, i. e. if value size is known
* statically, it is automatically accounted and shouldn't be specified by user. <p/> <p>If value size
* instead of this one. If value is a boxed primitive type, i. e. if value size is known
* statically, it is automatically accounted and shouldn't be specified by user. <p/> If value size
* varies moderately, specify the size higher than average, but lower than the maximum possible, to
* minimize average memory overuse. If value size varies in a wide range, it's better to use {@linkplain
* #entrySize(int) entry size} in "chunk" mode and configure it directly.
@@ -123,9 +123,9 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
/**
* Configures the constant number of bytes, taken by serialized form of values, put into maps, created by
* this builder. This is done by providing the {@code sampleValue}, all values should take the same number
* of bytes in serialized form, as this sample object. <p/> <p>If values are of boxed primitive type or
* of bytes in serialized form, as this sample object. If values are of boxed primitive type or
* {@link Byteable} subclass, i. e. if value size is known statically, it is automatically accounted and
* this method shouldn't be called. <p/> <p>If value size varies, method {@link #valueSize(int)} or {@link
* this method shouldn't be called. If value size varies, method {@link #valueSize(int)} or {@link
* #entrySize(int)} should be called instead of this one.
*
* @param sampleValue the sample value
@@ -149,7 +149,7 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
* // this class hasn't implemented yet, just for example
* .objectSerializer(new KryoObjectSerializer())
* .create();}</pre>
* <p/> <p>This serializer is used to serialize both keys and values, if they both require this: loosely
* <p/> This serializer is used to serialize both keys and values, if they both require this: loosely
* typed, nullable, and custom {@linkplain #keyMarshaller(BytesMarshaller) key} and {@linkplain
* #valueMarshaller(BytesMarshaller) value} marshallers are not configured.
*/
@@ -160,7 +160,7 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
}
/**
* {@inheritDoc} <p/> <p>If {@linkplain #valueMarshaller(BytesMarshaller) custom value marshaller} is
* {@inheritDoc} If {@linkplain #valueMarshaller(BytesMarshaller) custom value 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. Actually this is just a convenience method supporting value marshaller
src/main/java/net/openhft/chronicle/map/ReadValue.java
+3
View File
@@ -24,6 +24,9 @@ import org.jetbrains.annotations.NotNull;
* @param <RV> result type, map value type or Bytes
*/
interface ReadValue<RV> {
/**
* Updates the entry position by valueSize.
*/
RV readValue(@NotNull ThreadLocalCopies copies, Bytes entry, RV usingValue, long valueSize);
RV readNull();
}
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+104 -62
View File
@@ -48,36 +48,39 @@ import static net.openhft.lang.MemoryUnit.*;
import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
/**
* <h2>A Replicating Multi Master HashMap</h2> <p>Each remote hash map, mirrors its changes over to another
* remote hash map, neither hash map is considered the master store of data, each hash map uses timestamps to
* reconcile changes. We refer to an instance of a remote hash-map as a node. A node will be connected to any
* number of other nodes, for the first implementation the maximum number of nodes will be fixed. The data
* that is stored locally in each node will become eventually consistent. So changes made to one node, for
* example by calling put() will be replicated over to the other node. To achieve a high level of performance
* and throughput, the call to put() won’t block, with concurrentHashMap, It is typical to check the return
* code of some methods to obtain the old value for example remove(). Due to the loose coupling and lock free
* nature of this multi master implementation, this return value will only be the old value on the nodes
* local data store. In other words the nodes are only concurrent locally. Its worth realising that another
* node performing exactly the same operation may return a different value. However reconciliation will ensure
* the maps themselves become eventually consistent. </p> <h2>Reconciliation </h2> <p>If two ( or more nodes )
* were to receive a change to their maps for the same key but different values, say by a user of the maps,
* calling the put(key, value). Then, initially each node will update its local store and each local store
* will hold a different value, but the aim of multi master replication is to provide eventual consistency
* across the nodes. So, with multi master when ever a node is changed it will notify the other nodes of its
* change. We will refer to this notification as an event. The event will hold a timestamp indicating the time
* the change occurred, it will also hold the state transition, in this case it was a put with a key and
* value. Eventual consistency is achieved by looking at the timestamp from the remote node, if for a given
* key, the remote nodes timestamp is newer than the local nodes timestamp, then the event from the remote
* node will be applied to the local node, otherwise the event will be ignored. </p> <p>However there is an
* edge case that we have to concern ourselves with, If two nodes update their map at the same time with
* different values, we have to deterministically resolve which update wins, because of eventual consistency
* both nodes should end up locally holding the same data. Although it is rare two remote nodes could receive
* an update to their maps at exactly the same time for the same key, we have to handle this edge case, its
* therefore important not to rely on timestamps alone to reconcile the updates. Typically the update with the
* newest timestamp should win, but in this example both timestamps are the same, and the decision made to one
* node should be identical to the decision made to the other. We resolve this simple dilemma by using a node
* identifier, each node will have a unique identifier, the update from the node with the smallest identifier
* wins. </p>
* <h2>A Replicating Multi Master HashMap</h2> <p>Each remote hash map, mirrors its changes over to
* another remote hash map, neither hash map is considered the master store of data, each hash map
* uses timestamps to reconcile changes. We refer to an instance of a remote hash-map as a node. A
* node will be connected to any number of other nodes, for the first implementation the maximum
* number of nodes will be fixed. The data that is stored locally in each node will become
* eventually consistent. So changes made to one node, for example by calling put() will be
* replicated over to the other node. To achieve a high level of performance and throughput, the
* call to put() won’t block, with concurrentHashMap, It is typical to check the return code of some
* methods to obtain the old value for example remove(). Due to the loose coupling and lock free
* nature of this multi master implementation, this return value will only be the old value on the
* nodes local data store. In other words the nodes are only concurrent locally. Its worth realising
* that another node performing exactly the same operation may return a different value. However
* reconciliation will ensure the maps themselves become eventually consistent. </p>
* <h2>Reconciliation </h2> <p>If two ( or more nodes ) were to receive a change to their maps for
* the same key but different values, say by a user of the maps, calling the put(key, value). Then,
* initially each node will update its local store and each local store will hold a different value,
* but the aim of multi master replication is to provide eventual consistency across the nodes. So,
* with multi master when ever a node is changed it will notify the other nodes of its change. We
* will refer to this notification as an event. The event will hold a timestamp indicating the time
* the change occurred, it will also hold the state transition, in this case it was a put with a key
* and value. Eventual consistency is achieved by looking at the timestamp from the remote node, if
* for a given key, the remote nodes timestamp is newer than the local nodes timestamp, then the
* event from the remote node will be applied to the local node, otherwise the event will be
* ignored. </p> <p>However there is an edge case that we have to concern ourselves with, If two
* nodes update their map at the same time with different values, we have to deterministically
* resolve which update wins, because of eventual consistency both nodes should end up locally
* holding the same data. Although it is rare two remote nodes could receive an update to their maps
* at exactly the same time for the same key, we have to handle this edge case, its therefore
* important not to rely on timestamps alone to reconcile the updates. Typically the update with the
* newest timestamp should win, but in this example both timestamps are the same, and the decision
* made to one node should be identical to the decision made to the other. We resolve this simple
* dilemma by using a node identifier, each node will have a unique identifier, the update from the
* node with the smallest identifier wins. </p>
*
* @param <K> the entries key type
* @param <V> the entries value type
@@ -207,7 +210,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
BytesBytesInterop.INSTANCE, key, keySize, keyBytesToInstance,
getRemoteBytesValueInterops, value, valueBytesToInstance,
replaceIfPresent, readValue, false, localIdentifier, currentTime());;
replaceIfPresent, readValue, false, localIdentifier, currentTime());
;
}
/**
@@ -239,8 +243,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
* @param timeStamp the time that that change was made, this is used for replication
* @return the value that was replaced
*/
private V put0(K key, V value, boolean replaceIfPresent,
final byte identifier, long timeStamp) {
V put0(K key, V value, boolean replaceIfPresent,
final byte identifier, long timeStamp) {
checkKey(key);
checkValue(value);
ThreadLocalCopies copies = keyInteropProvider.getCopies(null);
@@ -254,6 +258,34 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
identifier, timeStamp);
}
Object remoteRemove(K key,
final byte remoteIdentifier, long timeStamp) {
checkKey(key);
K key1 = (K) key;
ThreadLocalCopies copies = keyInteropProvider.getCopies(null);
KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop);
copies = metaKeyInteropProvider.getCopies(copies);
MKI metaKeyInterop =
metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key1);
long keySize = metaKeyInterop.size(keyInterop, key1);
long hash = metaKeyInterop.hash(keyInterop, key1);
int segmentNum = getSegment(hash);
long segmentHash = segmentHash(hash);
ReplicatedChronicleMap.Segment segment = segment(segmentNum);
segment.writeLock(null);
try {
return segment.removeWithoutLock(copies, null, metaKeyInterop, keyInterop, key1, keySize,
keyIdentity(), this, (V) null, valueIdentity(), segmentHash, this, removeReturnsNull,
timeStamp, remoteIdentifier, true, false);
} finally {
segment.writeUnlock();
}
}
@Override
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<RV, ? super VBI>>
@@ -355,6 +387,11 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
}
@Override
void onRemotePut(VanillaChronicleMap<K, KI, MKI, V, VI, MVI>.Segment segment, long pos) {
onRelocation(segment, pos); // erase modification bits
}
@Override
void onRemove(VanillaChronicleMap<K, KI, MKI, V, VI, MVI>.Segment segment, long pos) {
for (long next = modIterSet.nextSetBit(0L); next > 0L;
@@ -367,6 +404,11 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
}
@Override
void onRemoteRemove(VanillaChronicleMap<K, KI, MKI, V, VI, MVI>.Segment segment, long pos) {
onRelocation(segment, pos); // erase modification bits
}
@Override
void onRelocation(VanillaChronicleMap<K, KI, MKI, V, VI, MVI>.Segment segment, long pos) {
for (long next = modIterSet.nextSetBit(0L); next > 0L;
@@ -380,8 +422,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* This method does not set a segment lock, A segment lock should be obtained before calling this method,
* especially when being used in a multi threaded context.
* This method does not set a segment lock, A segment lock should be obtained before calling
* this method, especially when being used in a multi threaded context.
*/
@Override
public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
@@ -454,8 +496,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* This method does not set a segment lock, A segment lock should be obtained before calling this method,
* especially when being used in a multi threaded context.
* This method does not set a segment lock, A segment lock should be obtained before calling
* this method, especially when being used in a multi threaded context.
*/
@Override
public void readExternalEntry(
@@ -471,7 +513,6 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
final byte remoteIdentifier;
if (id != 0) {
// isDeleted = false;
remoteIdentifier = id;
} else {
throw new IllegalStateException("identifier can't be 0");
@@ -900,8 +941,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
entry.writeBoolean(false);
// put callbacks
if (!remote)
onPut(this, localSegmentState.pos);
onPutMaybeRemote(localSegmentState.pos, remote);
if (bytesEventListener != null)
bytesEventListener.onPut(entry, 0L, metaDataBytes,
localSegmentState.valueSizePos, true);
@@ -1001,12 +1041,10 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* - if expectedValue is not null, returns Boolean.TRUE (removed) or
* Boolean.FALSE (entry not found), regardless the expectedValue object is Bytes instance
* (RPC call) or the value instance
* - if expectedValue is null:
* - if resultUnused is false, null or removed value is returned
* - if resultUnused is true, null is always returned
* - if expectedValue is not null, returns Boolean.TRUE (removed) or Boolean.FALSE (entry
* not found), regardless the expectedValue object is Bytes instance (RPC call) or the value
* instance - if expectedValue is null: - if resultUnused is false, null or removed value is
* returned - if resultUnused is true, null is always returned
*/
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
@@ -1057,6 +1095,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
entry.position(timestampPos);
entry.writeLong(timestamp);
entry.writeByte(identifier);
onRemoveMaybeRemote(pos, remote);
break; // to "key not found" location
}
@@ -1133,8 +1172,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
long timestampPos = entry.position();
if (shouldIgnore(entry, timestamp, identifier)) {
LOG.error("Trying to replace a value for key={} on the node with id={} " +
"at time={} (current time), but the entry is updated by node " +
"with id={} at time={}. Time is not monotonic across nodes!?",
"at time={} (current time), but the entry is updated by node " +
"with id={} at time={}. Time is not monotonic across nodes!?",
key, identifier, timestamp, entry.readByte(entry.position() - 1),
entry.readLong(entry.position() - ADDITIONAL_ENTRY_BYTES + 1));
return readValue.readNull();
@@ -1163,8 +1202,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* @param bootstrapOnlyLocalEntries if true - only entries that have been create from
* this node will be published during a bootstrap
* @param bootstrapOnlyLocalEntries if true - only entries that have been create from this
* node will be published during a bootstrap
*/
public void dirtyEntries(final long timeStamp,
final ModificationIterator.EntryModifiableCallback callback,
@@ -1300,14 +1339,15 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* <p>Once a change occurs to a map, map replication requires that these changes are picked up by another
* thread, this class provides an iterator like interface to poll for such changes. </p> <p>In most cases
* the thread that adds data to the node is unlikely to be the same thread that replicates the data over
* to the other nodes, so data will have to be marshaled between the main thread storing data to the map,
* and the thread running the replication. </p> <p>One way to perform this marshalling, would be to pipe
* the data into a queue. However, This class takes another approach. It uses a bit set, and marks bits
* which correspond to the indexes of the entries that have changed. It then provides an iterator like
* interface to poll for such changes. </p>
* <p>Once a change occurs to a map, map replication requires that these changes are picked up
* by another thread, this class provides an iterator like interface to poll for such changes.
* </p> <p>In most cases the thread that adds data to the node is unlikely to be the same thread
* that replicates the data over to the other nodes, so data will have to be marshaled between
* the main thread storing data to the map, and the thread running the replication. </p> <p>One
* way to perform this marshalling, would be to pipe the data into a queue. However, This class
* takes another approach. It uses a bit set, and marks bits which correspond to the indexes of
* the entries that have changed. It then provides an iterator like interface to poll for such
* changes. </p>
*
* @author Rob Austin.
*/
@@ -1344,7 +1384,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
*
* @param segmentIndex the index of the maps segment
* @param pos the position within this {@code segmentIndex}
* @return and index the has combined the {@code segmentIndex} and {@code pos} into a single value
* @return and index the has combined the {@code segmentIndex} and {@code pos} into a
* single value
*/
private long combine(int segmentIndex, long pos) {
return (((long) segmentIndex) << segmentIndexShift) | pos;
@@ -1368,8 +1409,9 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* you can continue to poll hasNext() until data becomes available. If are are in the middle of
* processing an entry via {@code nextEntry}, hasNext will return true until the bit is cleared
* you can continue to poll hasNext() until data becomes available. If are are in the middle
* of processing an entry via {@code nextEntry}, hasNext will return true until the bit is
* cleared
*
* @return true if there is an entry
*/
@@ -1420,7 +1462,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
// if the position was already cleared by another thread
// while we were trying to obtain segment lock (for example, in onReplication()),
// while we were trying to obtain segment lock (for example, in onRelocation()),
// go to pick up next (next iteration in while (true) loop)
} finally {
segment.readUnlock();
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+21 -11
View File
@@ -287,9 +287,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
this.connectionOutBuffer.clear();
// read a single byte back
if (!clientChannel.finishConnect() || !clientChannel.socket().isBound())
return;
// read a single byte back
while (this.connectionOutBuffer.position() <= 0) {
clientChannel.read(this.connectionOutBuffer);
clientChannel.read(this.connectionOutBuffer); // the remote identifier
checkTimeout(timeoutTime);
}
@@ -938,9 +941,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
// send out all the bytes
send(bytes, timeoutTime);
bytes.clear();
bytes.buffer().clear();
return blockingFetch(timeoutTime, transactionId);
} catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) {
checkTimeout(timeoutTime);
@@ -963,6 +963,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private Bytes blockingFetch(long timeoutTime, long transactionId) throws IOException {
bytes.clear();
bytes.buffer().clear();
// the number of bytes in the response
final int size = receive(SIZE_OF_SIZE, timeoutTime).readInt();
@@ -976,15 +979,22 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
} else
bytes.limit(bytes.capacity());
// block until we have received all the byte in this chunk
// block until we have received all the bytes in this chunk
receive(size, timeoutTime);
final boolean isException = bytes.readBoolean();
final long inTransactionId = bytes.readLong();
if (inTransactionId != transactionId)
throw new IllegalStateException("the received transaction-id=" + inTransactionId +
", does not match the expected transaction-id=" + transactionId);
final long inTransactionId = bytes.readLong();
final boolean isException = bytes.readBoolean();
if (inTransactionId != transactionId) {
LOG.error("", new IllegalStateException("Skipped Message with transaction-id=" +
inTransactionId +
", this can occur when you have another thread which has called the " +
"stateless client and terminated abruptly before the message has been " +
"returned from the server"));
blockingFetch(timeoutTime, transactionId);
}
if (isException) {
Throwable throwable = (Throwable) bytes.readObject();
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+26 -17
View File
@@ -203,6 +203,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
LOG.debug("", e);
} catch (Exception e) {
LOG.error("", e);
} catch (Throwable e) {
LOG.error("", e);
throw e;
} finally {
if (LOG.isDebugEnabled())
@@ -628,11 +631,14 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key);
try {
final int bytesJustWritten = attached.entryWriter.writeBufferToSocket(socketChannel,
final int len = attached.entryWriter.writeBufferToSocket(socketChannel,
approxTime);
if (bytesJustWritten > 0)
contemplateThrottleWrites(bytesJustWritten);
if (len == -1)
socketChannel.close();
if (len > 0)
contemplateThrottleWrites(len);
if (!attached.entryWriter.hasBytesToWrite() && !attached.entryWriter.isWorkIncomplete())
// TURN OP_WRITE_OFF
@@ -659,6 +665,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
int len = attached.entryReader.readSocketToBuffer(socketChannel);
if (len == -1) {
socketChannel.register(selector, 0);
if (replicationConfig.autoReconnectedUponDroppedConnection()) {
AbstractConnector connector = attached.connector;
if (connector != null)
@@ -1018,6 +1025,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
// if we still have some unwritten writer from last time
lastSentTime = approxTime;
assert in.position() <= Integer.MAX_VALUE;
int size = (int) in.position();
out.limit(size);
@@ -1027,7 +1035,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
LOG.debug("bytes-written=" + len);
if (len == size) {
// assert len == size;
out.clear();
in.clear();
} else {
@@ -1195,9 +1202,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
final long limit = out.limit();
out.limit(nextEntryPos);
boolean isStateless = (state != 1);
boolean isStatelessClient = (state != 1);
if (isStateless) {
if (isStatelessClient) {
if (statelessServerConnector == null) {
LOG.error("", new IllegalArgumentException("received an event " +
"from a stateless map, stateless maps are not " +
@@ -1247,6 +1254,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
return;
in.limit(in.position());
assert out.position() < Integer.MAX_VALUE;
in.position((int) out.position());
in.compact();
@@ -1268,8 +1276,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
return out.readLong();
else
return Long.MIN_VALUE;
// return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE;
}
public long remoteHeartbeatIntervalFromBuffer() {
@@ -1514,7 +1520,7 @@ class StatelessServerConnector<K, V> {
writeException(writer, e);
writeSizeAndFlags(sizeLocation, true, writer);
writeSizeAndFlags(sizeLocation+SIZE_OF_TRANSACTION_ID, true, writer);
return null;
}
@@ -1839,17 +1845,19 @@ class StatelessServerConnector<K, V> {
}
private long reflectTransactionId(Bytes reader, Bytes writer) {
final long sizeLocation = writer.position();
writer.skip(TRANSACTION_ID_OFFSET); // isException + SIZE_OF_SIZE
final long transactionId = reader.readLong();
final long sizeLocation = writer.position();
writer.skip(SIZE_OF_SIZE);
writer.writeLong(transactionId);
writer.writeBoolean(false); // isException
return sizeLocation;
}
private void writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) {
final long size = out.position() - locationOfSize;
out.writeInt(locationOfSize, (int) size); // size
out.writeBoolean(locationOfSize + SIZE_OF_SIZE, isException); // isException
out.skip(SIZE_OF_TRANSACTION_ID);
out.writeBoolean(isException); // isException
}
private void writeException(Bytes out, Throwable e) {
@@ -1881,10 +1889,11 @@ class StatelessServerConnector<K, V> {
private long header(Bytes writer, final long transactionId) {
final long sizeLocation = writer.position();
writer.skip(TRANSACTION_ID_OFFSET);
writer.skip(SIZE_OF_SIZE);
writer.writeLong(transactionId);
// exception
writer.writeLong(transactionId);
writer.skip(1);
// hasAnotherChunk
writer.skip(1);
@@ -1903,12 +1912,12 @@ class StatelessServerConnector<K, V> {
// size in bytes
writer.writeInt(size);
// is exception
writer.writeBoolean(false);
//transaction id;
writer.skip(8);
// is exception
writer.writeBoolean(false);
writer.writeBoolean(hasAnotherChunk);
// count
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+53 -28
View File
@@ -316,18 +316,18 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
}
final void checkKey(Object key) {
if (!kClass.isInstance(key)) {
// key.getClass will cause NPE exactly as needed
throw new ClassCastException("Key must be a " + kClass.getName() +
" but was a " + key.getClass());
}
// if (!kClass.isInstance(key)) {
// // key.getClass will cause NPE exactly as needed
// throw new ClassCastException("Key must be a " + kClass.getName() +
// " but was a " + key.getClass());
//}
}
final void checkValue(Object value) {
if (vClass != Void.class && !vClass.isInstance(value)) {
throw new ClassCastException("Value must be a " + vClass.getName() +
" but was a " + value.getClass());
}
// if (vClass != Void.class && !vClass.isInstance(value)) {
// throw new ClassCastException("Value must be a " + vClass.getName() +
// " but was a " + value.getClass());
// }
}
final void put(Bytes entry, Bytes output) {
@@ -384,7 +384,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
}
static abstract class ReadValueToBytes implements ReadValue<Bytes> {
private SizeMarshaller valueSizeMarshaller;
SizeMarshaller valueSizeMarshaller;
abstract Bytes bytes(long valueSize);
@@ -398,6 +398,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
usingBytes = bytes(valueSize);
valueSizeMarshaller.writeSize(usingBytes, valueSize);
usingBytes.write(entry, entry.position(), valueSize);
entry.skip(valueSize);
return usingBytes;
}
}
@@ -434,6 +435,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
@Override
Bytes bytes(long valueSize) {
valueSize = valueSizeMarshaller.sizeEncodingSize(valueSize) + valueSize;
if (lazyBytes != null) {
if (lazyBytes.capacity() < valueSize) {
DirectStore store = (DirectStore) lazyBytes.store();
@@ -1174,6 +1176,10 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
public final boolean replace(@net.openhft.lang.model.constraints.NotNull K key,
@net.openhft.lang.model.constraints.NotNull V oldValue,
@net.openhft.lang.model.constraints.NotNull V newValue) {
if (key == null || oldValue == null || newValue == null)
throw new NullPointerException();
checkValue(oldValue);
return (Boolean) replaceIfValueIs(key, oldValue, newValue);
}
@@ -1185,6 +1191,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
@SuppressWarnings("unchecked")
public final V replace(@net.openhft.lang.model.constraints.NotNull final K key,
@net.openhft.lang.model.constraints.NotNull final V value) {
if (key == null || value == null)
throw new NullPointerException();
return (V) replaceIfValueIs(key, null, value);
}
@@ -1359,10 +1368,18 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
// do nothing
}
void onRemoteRemove(Segment segment, long pos) {
// do nothing
}
void onPut(Segment segment, long pos) {
// do nothing
}
void onRemotePut(Segment segment, long pos) {
// do nothing
}
void onRelocation(Segment segment, long pos) {
// do nothing
}
@@ -1595,6 +1612,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
}
long startWriteLock = 0;
@Override
public final WriteLocked<K, KI, MKI, V, VI, MVI> writeLock(
@Nullable SegmentState segmentState) {
@@ -1930,8 +1948,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
metaValueInterop, valueInterop, value, valueSize, searchedHashLookup);
// put callbacks
if (!remote)
onPut(this, segmentState.pos);
onPutMaybeRemote(segmentState.pos, remote);
if (bytesEventListener != null)
bytesEventListener.onPut(entry, 0L, metaDataBytes, valueSizePos, false);
if (eventListener != null) {
@@ -1942,6 +1959,14 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return resultUnused ? null : prevValue;
}
final void onPutMaybeRemote(long pos, boolean remote) {
if (remote) {
onRemotePut(this, pos);
} else {
onPut(this, pos);
}
}
/**
* Returns value size, writes the entry (key, value, sizes) to the entry, after this method
* call entry positioned after value bytes written (i. e. at the end of entry), sets entry
@@ -2021,10 +2046,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return ret;
}
private void setBlocks(long pos, int blocks) {
freeList.set(pos, pos + blocks);
}
private void updateNextPosToSearchFrom(long allocated, int blocks) {
if ((nextPosToSearchFrom = allocated + blocks) >= entriesPerSegment)
nextPosToSearchFrom = 0L;
@@ -2155,16 +2176,15 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
// update segment state
if (removeFromMultiMap) {
hashLookup.removePrevPos();
long entrySizeInBytes = entry.positionAddr() + valueSize - entry.startAddr();
free(pos, inBlocks(entrySizeInBytes));
} else {
hashLookup.removePosition(pos);
}
decrementSize();
long entrySizeInBytes = entry.positionAddr() + valueSize - entry.startAddr();
free(pos, inBlocks(entrySizeInBytes));
// remove callbacks
if (!remote)
onRemove(this, pos);
onRemoveMaybeRemote(pos, remote);
if (bytesEventListener != null)
bytesEventListener.onRemove(entry, 0L, metaDataBytes, valueSizePos);
if (eventListener != null) {
@@ -2177,6 +2197,14 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return booleanResult ? Boolean.TRUE : removedValue;
}
final void onRemoveMaybeRemote(long pos, boolean remote) {
if (remote) {
onRemoteRemove(this, pos);
} else {
onRemove(this, pos);
}
}
private <KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>>
boolean containsKey(ThreadLocalCopies copies,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, long hash2) {
@@ -2331,15 +2359,15 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
MetaBytesWriter<E, ? super EW> metaElemWriter, EW elemWriter, E newElem,
long newElemSize,
MultiMap searchedHashLookup) {
long valueSizeAddr = entry.address() + valueSizePos;
long entryStartAddr = entry.address();
long valueSizeAddr = entryStartAddr + valueSizePos;
long newValueAddr = alignment.alignAddr(
valueSizeAddr + valueSizeMarshaller.sizeEncodingSize(newElemSize));
long newEntryEndAddr = newValueAddr + newElemSize;
long entryStartAddr = entry.address();
long oldEntrySize = entryEndAddr - entryStartAddr;
int oldSizeInBlocks = inBlocks(oldEntrySize);
newValueDoesNotFit:
if (newEntryEndAddr != entryEndAddr) {
long oldEntrySize = entryEndAddr - entryStartAddr;
int oldSizeInBlocks = inBlocks(oldEntrySize);
int newSizeInBlocks = inBlocks(newEntryEndAddr - entryStartAddr);
if (newSizeInBlocks > oldSizeInBlocks) {
if (newSizeInBlocks > MAX_ENTRY_OVERSIZE_FACTOR) {
@@ -2373,9 +2401,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
// But if these blocks will be taken by that time,
// this entry will need to be relocated.
}
} else {
// todo investivate when we are able to avoid this assignment
setBlocks(pos, oldSizeInBlocks);
}
// Common code for all cases
entry.position(valueSizePos);
@@ -2585,11 +2610,11 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
final long entryEndAddr = entry.positionAddr() + valueSize;
if (removeFromMultiMap) {
segment.hashLookup().remove(segmentHash, pos);
segment.free(pos, segment.inBlocks(entryEndAddr - entry.address()));
} else {
segment.hashLookup().removePosition(pos);
}
segment.decrementSize();
segment.free(pos, segment.inBlocks(entryEndAddr - entry.address()));
// remove callbacks
onRemove(segment, pos);
src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
+2 -2
View File
@@ -45,12 +45,12 @@ import java.util.concurrent.TimeUnit;
* {@code ChronicleSetBuilder} manages the whole set of {@link ChronicleSet} configurations, could
* be used as a classic builder and/or factory.
*
* <p>{@code ChronicleMapOnHeapUpdatableBuilder} is mutable, see a note in {@link
* <p>{@code ChronicleMapBuilder} is mutable, see a note in {@link
* ChronicleHashBuilder} interface documentation.
*
* @param <E> element type of the sets, created by this builder
* @see ChronicleSet
* @see net.openhft.chronicle.map.OnHeapUpdatableChronicleMapBuilder
* @see net.openhft.chronicle.map.ChronicleMapBuilder
*/
public class ChronicleSetBuilder<E>
implements ChronicleHashBuilder<E, ChronicleSet<E>, ChronicleSetBuilder<E>> {
src/test/java/eg/BGChronicleTest.java
+242
View File
@@ -0,0 +1,242 @@
package eg;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Random;
import java.util.concurrent.TimeUnit;
public class BGChronicleTest {
private static final int CHRONICLE_HEARTBEAT_SECONDS = 5;
public static final long MAX_NUMBER_OF_EATERIES = 3000000L;
public static final String DEFAULT_HOST = "localhost";
public static final int DEFAULT_PORT = 9050;
Random rnd = new Random(System.currentTimeMillis());
byte[] key = new byte[17];
byte[] value = new byte[11];
public ChronicleMap<byte[], byte[]> startChronicleMapServer(int port) throws InterruptedException, IOException {
TcpTransportAndNetworkConfig serverConfig = TcpTransportAndNetworkConfig.of(port)
.heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS).name("serverMap")
.autoReconnectedUponDroppedConnection(true);
return ChronicleMapBuilder.of(byte[].class, byte[].class)
.entries(MAX_NUMBER_OF_EATERIES)
.keySize(17)
.valueSize(11)
.putReturnsNull(true)
.removeReturnsNull(true)
.replication((byte) 1, serverConfig)
.create();
}
/**
* over loop-back on my mac I get around 50,000 messages per second of my local network ( LAN ),
* I get around 730 messages per second
*
* @param hostNameOrIpAddress
* @param port
* @param numberOfMessages
* @throws IOException
*/
public void startChronicleMapClient(String hostNameOrIpAddress, final int port, final long
numberOfMessages) throws IOException {
System.out.println("client starting");
try (ChronicleMap<byte[], byte[]> map = ChronicleMapBuilder.of(byte[].class, byte[].class)
.keySize(17)
.valueSize(11)
.putReturnsNull(true)
.removeReturnsNull(true)
.statelessClient(new InetSocketAddress(hostNameOrIpAddress, port)).create()) {
System.out.println("client started");
long start = System.nanoTime();
for (long count = 0; count < numberOfMessages; count++) {
if (rnd.nextBoolean()) {
// puts
rnd.nextBytes(key);
rnd.nextBytes(value);
map.put(key, value);
} else {
//gets
rnd.nextBytes(key);
map.get(key);
}
if (count % 50000 == 0) {
logMessagesPerSecond(count, start);
}
}
logMessagesPerSecond(numberOfMessages, start);
}
}
/**
* over loop-back on my mac I get around 50,000 messages per second of my local network ( LAN ),
* I get around 730 messages per second
*
* @param hostNameOrIpAddress
* @param port
* @param numberOfMessages
* @throws IOException
*/
public void startReplicatingChronicleMapClient(String hostNameOrIpAddress, final int port,
final long numberOfMessages) throws
IOException {
System.out.println("client starting");
final InetSocketAddress serverHostPort = new InetSocketAddress(hostNameOrIpAddress, port);
TcpTransportAndNetworkConfig clientConfig = TcpTransportAndNetworkConfig.of(port + 1, serverHostPort)
.heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS).name("clientMap")
.autoReconnectedUponDroppedConnection(true);
try (ChronicleMap<byte[], byte[]> map = ChronicleMapBuilder.of(byte[].class, byte[].class)
.entries(MAX_NUMBER_OF_EATERIES)
.keySize(17)
.valueSize(11)
.putReturnsNull(true)
.removeReturnsNull(true)
.replication((byte) 2, clientConfig)
.create()) {
System.out.println("client started");
long start = System.nanoTime();
for (long count = 0; count < numberOfMessages; count++) {
if (rnd.nextBoolean()) {
// puts
rnd.nextBytes(key);
rnd.nextBytes(value);
map.put(key, value);
} else {
//gets
rnd.nextBytes(key);
map.get(key);
}
if (count % 50000 == 0) {
logMessagesPerSecond(count, start);
}
}
logMessagesPerSecond(numberOfMessages, start);
}
}
private void logMessagesPerSecond(double numberOfMessages, long start) {
double messagesPerSecond = numberOfMessages / (double) TimeUnit.NANOSECONDS
.toSeconds(System.nanoTime() - start);
System.out.println("messages per seconds " + messagesPerSecond);
}
/**
* for loop back args = []
*
* for server - eg.BGChronicleTest server for client - eg.BGChronicleTest client
*
* @param args
* @throws IOException
* @throws InterruptedException
*/
public static void main(String... args) throws IOException, InterruptedException {
final BGChronicleTest o = new BGChronicleTest();
ChronicleMap<byte[], byte[]> serverMap = null;
try {
// loop-back
if (args.length == 0) {
serverMap = o.startChronicleMapServer(DEFAULT_PORT);
o.startChronicleMapClient(DEFAULT_HOST, DEFAULT_PORT, MAX_NUMBER_OF_EATERIES);
} else {
// server
if (args.length > 0 && args[0].equalsIgnoreCase("server")) {
int port = getPort(args, 1);
serverMap = o.startChronicleMapServer(port);
System.out.println("server ON");
} else
System.out.println("server if OFF");
// client
if (args.length == 0 || (args.length > 1 && args[0].equalsIgnoreCase("client"))) {
// stateless replicating client
String host = getHost(args, 1);
int port = getPort(args, 2);
System.out.println("client ON....");
o.startChronicleMapClient(host, port, MAX_NUMBER_OF_EATERIES);
// replicating client
} else if (args.length > 1 && args[0].equalsIgnoreCase("replicatingClient")) {
String host = getHost(args, 1);
int port = getPort(args, 2);
System.out.println("client ON....");
o.startReplicatingChronicleMapClient(host, port, MAX_NUMBER_OF_EATERIES);
} else {
System.out.println("client if OFF");
for (; ; ) {
Thread.sleep(1000);
}
}
System.out.println("client if OFF");
for (; ; ) {
Thread.sleep(1000);
}
}
} finally {
if (serverMap != null)
serverMap.close();
}
}
private static String getHost(String[] args, int index) {
String host = args.length > index && args[index] != null ? args[index] : DEFAULT_HOST;
System.out.println(host);
return host;
}
private static int getPort(String[] args, int index) {
int port = args.length > index && args[index] != null ? Integer.valueOf(args[index]) :
DEFAULT_PORT;
System.out.println(port);
return port;
}
public void test() throws IOException, InterruptedException {
main();
}
}
src/test/java/net/openhft/chronicle/map/TestBooleanValues.java → src/test/java/net/openhft/chronicle/map/BooleanValuesTest.java
+1 -1
View File
@@ -26,7 +26,7 @@ import java.io.IOException;
/**
* @author Rob Austin.
*/
public class TestBooleanValues {
public class BooleanValuesTest {
/**
* see issue http://stackoverflow.com/questions/26219313/strange-npe-from-chronicle-map-toy-code
src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java
+39 -16
View File
@@ -20,10 +20,11 @@ import static org.junit.Assert.*;
@Ignore
public class CHMUseCasesTest {
/**
* String is not as efficient as CharSequence as a key or value but easier to use The key can only be on heap and
* variable length serialised.
* String is not as efficient as CharSequence as a key or value but easier to use The key can
* only be on heap and variable length serialised.
*/
@Test
@Ignore("Not yet implemented")
public void testStringStringMap() throws ExecutionException, InterruptedException {
/* TODO run the same test for multiple types of map stores
for(ChronicleMapBuilder<String, String> chmb : Arrays.asList(
@@ -72,10 +73,11 @@ public class CHMUseCasesTest {
}
/**
* CharSequence is more efficient when object creation is avoided. The key can only be on heap and variable length
* serialised.
* CharSequence is more efficient when object creation is avoided. The key can only be on heap
* and variable length serialised.
*/
@Test
@Ignore("Not yet implemented")
public void testCharSequenceCharSequenceMap() throws ExecutionException, InterruptedException {
try (ChronicleMap<CharSequence, CharSequence> map = ChronicleMapBuilder
.of(CharSequence.class, CharSequence.class)
@@ -138,6 +140,7 @@ public class CHMUseCasesTest {
* StringValue represents any bean which contains a String Value
*/
@Test
@Ignore("Not yet implemented")
public void testStringValueStringValueMap() {
try (ChronicleMap<StringValue, StringValue> map = ChronicleMapBuilder
.of(StringValue.class, StringValue.class)
@@ -229,6 +232,7 @@ public class CHMUseCasesTest {
}
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testIntegerIntegerMap() throws ExecutionException, InterruptedException {
try (ChronicleMap<Integer, Integer> map = ChronicleMapBuilder
.of(Integer.class, Integer.class)
@@ -287,6 +291,7 @@ public class CHMUseCasesTest {
}
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testLongLongMap() throws ExecutionException, InterruptedException {
try (ChronicleMap<Long, Long> map = ChronicleMapBuilder
.of(Long.class, Long.class)
@@ -334,6 +339,7 @@ public class CHMUseCasesTest {
}
@Test
@Ignore("Not yet implemented")
public void testDoubleDoubleMap() throws ExecutionException, InterruptedException {
try (ChronicleMap<Double, Double> map = ChronicleMapBuilder
.of(Double.class, Double.class)
@@ -381,6 +387,7 @@ public class CHMUseCasesTest {
}
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testByteArrayByteArrayMap() throws ExecutionException, InterruptedException {
try (ChronicleMap<byte[], byte[]> map = ChronicleMapBuilder
.of(byte[].class, byte[].class)
@@ -431,6 +438,7 @@ public class CHMUseCasesTest {
}
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testByteBufferByteBufferMap() throws ExecutionException, InterruptedException {
try (ChronicleMap<ByteBuffer, ByteBuffer> map = ChronicleMapBuilder
.of(ByteBuffer.class, ByteBuffer.class)
@@ -529,6 +537,7 @@ public class CHMUseCasesTest {
}
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testByteBufferDirectByteBufferMap() throws ExecutionException, InterruptedException {
try (ChronicleMap<ByteBuffer, ByteBuffer> map = ChronicleMapBuilder
.of(ByteBuffer.class, ByteBuffer.class)
@@ -589,6 +598,7 @@ public class CHMUseCasesTest {
}
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testIntValueIntValueMap() {
try (ChronicleMap<IntValue, IntValue> map = ChronicleMapBuilder
.of(IntValue.class, IntValue.class)
@@ -671,6 +681,7 @@ public class CHMUseCasesTest {
* For unsigned int -> unsigned int entries, the key can be on heap or off heap.
*/
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testUnsignedIntValueUnsignedIntValueMap() {
try (ChronicleMap<UnsignedIntValue, UnsignedIntValue> map = ChronicleMapBuilder
.of(UnsignedIntValue.class, UnsignedIntValue.class)
@@ -753,6 +764,7 @@ public class CHMUseCasesTest {
* For int values, the key can be on heap or off heap.
*/
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testIntValueShortValueMap() {
try (ChronicleMap<IntValue, ShortValue> map = ChronicleMapBuilder
.of(IntValue.class, ShortValue.class)
@@ -835,6 +847,7 @@ public class CHMUseCasesTest {
* For int -> unsigned short values, the key can be on heap or off heap.
*/
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testIntValueUnsignedShortValueMap() {
try (ChronicleMap<IntValue, UnsignedShortValue> map = ChronicleMapBuilder
.of(IntValue.class, UnsignedShortValue.class)
@@ -917,6 +930,7 @@ public class CHMUseCasesTest {
* For int values, the key can be on heap or off heap.
*/
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testIntValueCharValueMap() {
try (ChronicleMap<IntValue, CharValue> map = ChronicleMapBuilder
.of(IntValue.class, CharValue.class)
@@ -999,6 +1013,7 @@ public class CHMUseCasesTest {
* For int-> byte entries, the key can be on heap or off heap.
*/
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testIntValueUnsignedByteMap() {
try (ChronicleMap<IntValue, UnsignedByteValue> map = ChronicleMapBuilder
.of(IntValue.class, UnsignedByteValue.class)
@@ -1081,6 +1096,7 @@ public class CHMUseCasesTest {
* For int values, the key can be on heap or off heap.
*/
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testIntValueBooleanValueMap() {
try (ChronicleMap<IntValue, BooleanValue> map = ChronicleMapBuilder
.of(IntValue.class, BooleanValue.class)
@@ -1163,6 +1179,7 @@ public class CHMUseCasesTest {
* For float values, the key can be on heap or off heap.
*/
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testFloatValueFloatValueMap() {
try (ChronicleMap<FloatValue, FloatValue> map = ChronicleMapBuilder
.of(FloatValue.class, FloatValue.class)
@@ -1245,18 +1262,24 @@ public class CHMUseCasesTest {
* For double values, the key can be on heap or off heap.
*/
@Test
@Ignore("HCOLL-225 Add support for heap and native object in the same map")
public void testDoubleValueDoubleValueMap() {
try (ChronicleMap<DoubleValue, DoubleValue> map = ChronicleMapBuilder
.of(DoubleValue.class, DoubleValue.class)
.entrySize(16)
.create()) {
DoubleValue key1 = DataValueClasses.newDirectInstance(DoubleValue.class);
DoubleValue key2 = DataValueClasses.newInstance(DoubleValue.class);
DoubleValue value1 = DataValueClasses.newDirectInstance(DoubleValue.class);
DoubleValue value2 = DataValueClasses.newInstance(DoubleValue.class);
key1.setValue(1);
value1.setValue(11);
assertEquals(null, map.get(key1));
map.put(key1, value1);
assertEquals(value1, map.get(key1));
key2.setValue(2);
@@ -1325,6 +1348,7 @@ public class CHMUseCasesTest {
* For long values, the key can be on heap or off heap.
*/
@Test
@Ignore("HCOLL-226 implement disableOversizedEntries()")
public void testLongValueLongValueMap() {
try (ChronicleMap<LongValue, LongValue> map = ChronicleMapBuilder
.of(LongValue.class, LongValue.class)
@@ -1416,30 +1440,29 @@ public class CHMUseCasesTest {
.valueMarshaller(ListMarshaller.of(new StringMarshaller(8)))
.create()) {
map.put("1", Collections.<String>emptyList());
map.put("2", Arrays.asList("one"));
map.put("2", Arrays.asList("two-A"));
List<String> list1 = new ArrayList<>();
try (WriteContext wc = map.acquireUsingLocked("1", list1)) {
list1.add("two");
assertEquals(Arrays.asList("two"), list1);
assertEquals(Collections.EMPTY_LIST, list1);
}
List<String> list2 = new ArrayList<>();
try (ReadContext rc = map.getUsingLocked("1", list2)) {
assertTrue(rc.present());
assertEquals(Arrays.asList("two"), list2);
}
try (WriteContext wc = map.acquireUsingLocked("2", list1)) {
list1.add("three");
assertEquals(Arrays.asList("three"), list1);
}
try (ReadContext rc = map.getUsingLocked("2", list2)) {
assertTrue(rc.present());
assertEquals(Arrays.asList("one", "three"), list2);
list2.add("two-B"); // this is not written as it only a read context
assertEquals(Arrays.asList("two-A", "two-B"), list2);
}
try (ReadContext rc = map.getUsingLocked("2", list2)) {
assertTrue(rc.present());
assertEquals(Arrays.asList("two-A"), list2);
}
}
}
@Test
@Ignore("Fix LinkedHashSet Marshaller")
public void testSetValue() {
try (ChronicleMap<String, Set<String>> map = ChronicleMapBuilder
.of(String.class, (Class<Set<String>>) (Class) Set.class)
src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java
+3 -3
View File
@@ -551,7 +551,8 @@ public class ChronicleMapTest {
threads[t].join();
}
assertEquals(noOfThreads * iterations, map.acquireUsing(key, new LongValue$$Native()).getValue());
assertEquals(noOfThreads * iterations,
map.acquireUsing(key, new LongValue$$Native()).getValue());
map.close();
}
@@ -1398,8 +1399,7 @@ public class ChronicleMapTest {
}
}
@Ignore("JIRA raised for failing test HCOLL-222")
@Test
@Test(expected = NullPointerException.class)
public void testPutLongValue() throws IOException {
final ChronicleMapBuilder<CharSequence, LongValue> builder = ChronicleMapBuilder
.of(CharSequence.class, LongValue.class)
src/test/java/net/openhft/chronicle/map/ConstantSizeBySampleTest.java
+2 -3
View File
@@ -106,14 +106,13 @@ public class ConstantSizeBySampleTest {
@Override
public boolean equals(Object obj) {
if (!(obj instanceof ExternalizableData))
if (!(obj instanceof SerializableData))
return false;
return Arrays.equals(((ExternalizableData) obj).data, data);
return Arrays.equals(((SerializableData) obj).data, data);
}
}
@Test
@Ignore
public void testUnexpectedlyLongConstantSerializableValues() throws IOException {
try ( ChronicleMap<Long, SerializableData> map =
ChronicleMapBuilder.of(Long.class, SerializableData.class)
src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java → src/test/java/net/openhft/chronicle/map/EventListenerWithTCPSocketReplicationTest.java
+1 -2
View File
@@ -35,7 +35,7 @@ import java.util.concurrent.atomic.AtomicReference;
*
* @author Rob Austin.
*/
public class EventListenerTestWithTCPSocketReplication {
public class EventListenerWithTCPSocketReplicationTest {
ChronicleMap<Integer, CharSequence> map1;
ChronicleMap<Integer, CharSequence> map2;
@@ -180,7 +180,6 @@ public class EventListenerTestWithTCPSocketReplication {
}
@Test
@Ignore("TODO Failing test")
public void testRemoteNotifyRemove() throws IOException, InterruptedException {
// ---------- SERVER1 1 ----------
src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java
+208
View File
@@ -0,0 +1,208 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.TimeProvider;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.model.Byteable;
import net.openhft.lang.model.DataValueClasses;
import net.openhft.lang.values.IntValue;
import org.junit.*;
import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author Rob Austin.
*/
public class NoTcpReplicationSoakTest {
private ReplicatedChronicleMap<Integer, ?, ?, CharSequence, ?, ?> map1;
private ReplicatedChronicleMap<Integer, ?, ?, CharSequence, ?, ?> map2;
private IntValue value;
static int s_port = 8010;
private long time;
TimeProvider timeProvider = new TimeProvider() {
@Override
public long currentTimeMillis() {
return time;
}
};
@Before
public void setup() throws IOException {
value = DataValueClasses.newDirectReference(IntValue.class);
((Byteable) value).bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0);
{
map1 = (ReplicatedChronicleMap) ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.name("map1")
.replication((byte) 1)
.timeProvider(timeProvider)
.create();
}
{
map2 = (ReplicatedChronicleMap) ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.name("map2")
.replication((byte) 2)
.timeProvider(timeProvider)
.create();
}
s_port += 2;
}
@After
public void tearDown() throws InterruptedException {
for (final Closeable closeable : new Closeable[]{map1, map2}) {
try {
closeable.close();
} catch (Exception e) {
e.printStackTrace();
}
}
System.gc();
}
Set<Thread> threads;
@Before
public void sampleThreads() {
threads = Thread.getAllStackTraces().keySet();
}
@After
public void checkThreadsShutdown() {
StatelessClientTest.checkThreadsShutdown(threads);
}
AtomicInteger task = new AtomicInteger();
@Test
@Ignore
public void testSoakTestWithRandomData() throws IOException, InterruptedException {
ExecutorService executorService = Executors.newSingleThreadExecutor();
System.out.print("SoakTesting ");
for (int j = 1; j < 10000; j++) {
if (j % 100 == 0)
System.out.print(".");
Random rnd = new Random(j);
if (rnd.nextInt(10) < 2)
time++;
final long t = time;
final int key = rnd.nextInt(1000);
// add random delay
addRandomDelay(executorService, rnd);
if (rnd.nextBoolean()) {
final CharSequence value = "test" + j;
if (rnd.nextBoolean()) {
map1.put(key, value);
task.incrementAndGet();
executorService.submit(new Runnable() {
@Override
public void run() {
map2.put0(key, value, true, (byte) 1, t);
task.decrementAndGet();
}
});
} else {
map2.put(key, value);
task.incrementAndGet();
executorService.submit(new Runnable() {
@Override
public void run() {
map1.put0(key, value, true, (byte) 2, t);
task.decrementAndGet();
}
});
}
} else {
if (rnd.nextBoolean()) {
map1.remove(key);
task.incrementAndGet();
executorService.submit(new Runnable() {
@Override
public void run() {
map2.remoteRemove(key, (byte) 1, t);
task.decrementAndGet();
}
});
} else {
map2.remove(key);
task.incrementAndGet();
executorService.submit(new Runnable() {
@Override
public void run() {
map1.remoteRemove(key, (byte) 2, t);
task.decrementAndGet();
}
});
}
}
}
while (task.get() > 0) {
}
Assert.assertEquals(new TreeMap(map1), new TreeMap(map2));
}
private void addRandomDelay(ExecutorService executorService, Random rnd) {
if (rnd.nextInt(30) < 2) {
task.incrementAndGet();
executorService.submit(new Runnable() {
@Override
public void run() {
task.decrementAndGet();
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
}
src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java
+1 -2
View File
@@ -161,14 +161,13 @@ public class TCPSocketReplicationIntValueTest {
for (; t < timeOutMs; t++) {
if (map1.equals(map2))
break;
Thread.sleep(25);
Thread.sleep(1);
}
assertEquals(map1, map2);
}
// TODO test this with larger sizes.
@Test(timeout = 12000)
@Ignore("Doesn't work, maps need to check equality.")
public void testSoakTestWithRandomData() throws IOException, InterruptedException {
final long start = System.currentTimeMillis();
System.out.print("SoakTesting ");
src/test/java/net/openhft/chronicle/map/TCPSocketReplicationPostConnectionTest.java
+2 -2
View File
@@ -61,12 +61,12 @@ public class TCPSocketReplicationPostConnectionTest {
s_port += 2;
map1 = TCPSocketReplication4WayMapTest.newTcpSocketShmIntString((byte) 1, port);
map1.put(5, "EXAMPLE-2");
Thread.sleep(1);
Thread.sleep(10);
map2 = TCPSocketReplication4WayMapTest.newTcpSocketShmIntString((byte) 2, port + 1,
new InetSocketAddress("localhost", port));
// allow time for the recompilation to resolve
waitTillEqual(500);
waitTillEqual(15000);
assertEquals(map1, map2);
assertTrue(!map1.isEmpty());
src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java
+42 -3
View File
@@ -28,7 +28,6 @@ import java.net.InetSocketAddress;
import java.util.Random;
import java.util.Set;
import static net.openhft.chronicle.map.Builder.getPersistenceFile;
import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder;
import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmIntString;
import static org.junit.Assert.assertEquals;
@@ -52,7 +51,7 @@ public class TCPSocketReplicationTest {
ChronicleMapBuilder<Integer, CharSequence> map1Builder =
newTcpSocketShmBuilder(Integer.class, CharSequence.class,
(byte) 1, port, new InetSocketAddress("localhost", port + 1));
map1 = map1Builder.createPersistedTo(getPersistenceFile());
map1 = map1Builder.create();
map2 = newTcpSocketShmIntString((byte) 2, port + 1);
s_port += 2;
}
@@ -116,6 +115,46 @@ public class TCPSocketReplicationTest {
assertTrue(!map1.isEmpty());
}
@Test
public void test2() throws IOException, InterruptedException {
map2.put(1, "EXAMPLE-1");
map2.remove(1);
map1.put(1, "EXAMPLE-2");
map1.remove(1);
map2.put(1, "EXAMPLE-1");
// allow time for the recompilation to resolve
waitTillEqual(5000);
assertEquals(map1, map2);
assertTrue(!map1.isEmpty());
}
@Test
public void test4() throws IOException, InterruptedException {
map2.put(1, "EXAMPLE-1");
map2.remove(1);
map1.put(1, "EXAMPLE-2");
map1.remove(1);
map2.put(1, "EXAMPLE-1");
map2.remove(1);
// allow time for the recompilation to resolve
waitTillEqual(5000);
assertEquals(map1, map2);
assertTrue(map1.isEmpty());
}
@Test
public void testBufferOverflow() throws IOException, InterruptedException {
@@ -168,7 +207,7 @@ public class TCPSocketReplicationTest {
}
}
System.out.println("");
waitTillEqual(10000);
waitTillEqual(1000);
}
}
src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java
+146
View File
@@ -0,0 +1,146 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.model.Byteable;
import net.openhft.lang.model.DataValueClasses;
import net.openhft.lang.values.IntValue;
import org.junit.*;
import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.TimeUnit;
/**
* @author Rob Austin.
*/
public class TcpReplicationSoakTest {
private ChronicleMap<Integer, CharSequence> map1;
private ChronicleMap<Integer, CharSequence> map2;
private IntValue value;
static int s_port = 8010;
@Before
public void setup() throws IOException {
value = DataValueClasses.newDirectReference(IntValue.class);
((Byteable) value).bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0);
final InetSocketAddress endpoint = new InetSocketAddress("localhost", s_port + 1);
{
final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port,
endpoint).autoReconnectedUponDroppedConnection(true).name(" map1")
.heartBeatInterval(1, TimeUnit.SECONDS);
map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(Builder.SIZE + Builder.SIZE)
.actualSegments(1)
.name("map1")
.replication((byte) 1, tcpConfig1)
.create();
}
{
final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of
(s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2")
.heartBeatInterval(1, TimeUnit.SECONDS);
map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(Builder.SIZE + Builder.SIZE)
.name("map2")
.replication((byte) 2, tcpConfig2)
.create();
}
s_port += 2;
}
@After
public void tearDown() throws InterruptedException {
for (final Closeable closeable : new Closeable[]{map1, map2}) {
try {
closeable.close();
} catch (Exception e) {
e.printStackTrace();
}
}
System.gc();
}
Set<Thread> threads;
@Before
public void sampleThreads() {
threads = Thread.getAllStackTraces().keySet();
}
@After
public void checkThreadsShutdown() {
StatelessClientTest.checkThreadsShutdown(threads);
}
@Test
public void testSoakTestWithRandomData() throws IOException, InterruptedException {
System.out.print("SoakTesting ");
for (int j = 1; j < 2 * Builder.SIZE; j++) {
if (j % 100 == 0)
System.out.print(".");
Random rnd = new Random(j);
for (int i = 1; i < 10; i++) {
final int select = rnd.nextInt(2);
final ChronicleMap<Integer, CharSequence> map = select > 0 ? map1 : map2;
if (rnd.nextBoolean()) {
map.put(rnd.nextInt(Builder.SIZE), "test" + j);
} else {
map.remove(rnd.nextInt(Builder.SIZE));
}
}
}
Thread.sleep(1000);
System.out.println("\nwaiting till equal");
waitTillEqual(10000);
Assert.assertEquals(new TreeMap(map1), new TreeMap(map2));
}
private void waitTillEqual(final int timeOutMs) throws InterruptedException {
Map map1UnChanged = new HashMap();
Map map2UnChanged = new HashMap();
int numberOfTimesTheSame = 0;
for (int t = 0; t < timeOutMs + 100; t++) {
if (map1.equals(map2)) {
if (map1.equals(map1UnChanged) && map2.equals(map2UnChanged)) {
numberOfTimesTheSame++;
} else {
numberOfTimesTheSame = 0;
map1UnChanged = new HashMap(map1);
map2UnChanged = new HashMap(map2);
}
Thread.sleep(1);
if (numberOfTimesTheSame == 100) {
System.out.println("same");
break;
}
}
Thread.sleep(1);
}
}
}
src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java
+166
View File
@@ -0,0 +1,166 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.TimeProvider;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.model.Byteable;
import net.openhft.lang.model.DataValueClasses;
import net.openhft.lang.values.IntValue;
import org.junit.*;
import org.mockito.Mockito;
import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.TimeUnit;
/**
* @author Rob Austin.
*/
public class TcpTimeBasedReplicationSoakTest {
private ChronicleMap<Integer, CharSequence> map1;
private ChronicleMap<Integer, CharSequence> map2;
private IntValue value;
static int s_port = 8010;
private TimeProvider timeProvider;
int t = 0;
@Before
public void setup() throws IOException {
value = DataValueClasses.newDirectReference(IntValue.class);
((Byteable) value).bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0);
final InetSocketAddress endpoint = new InetSocketAddress("localhost", s_port + 1);
timeProvider = new TimeProvider() {
Random rnd = new Random(4);
@Override
public long currentTimeMillis() {
if (rnd.nextBoolean())
return t++;
else
return t;
}
};
{
final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port,
endpoint);
map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(Builder.SIZE)
.timeProvider(timeProvider)
.replication((byte) 1, tcpConfig1)
.create();
}
{
final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of(s_port + 1);
map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(Builder.SIZE)
.timeProvider(timeProvider)
.replication((byte) 2, tcpConfig2)
.create();
}
s_port += 2;
}
@After
public void tearDown() throws InterruptedException {
for (final Closeable closeable : new Closeable[]{map1, map2}) {
try {
closeable.close();
} catch (Exception e) {
e.printStackTrace();
}
}
System.gc();
}
Set<Thread> threads;
@Before
public void sampleThreads() {
threads = Thread.getAllStackTraces().keySet();
}
@After
public void checkThreadsShutdown() {
StatelessClientTest.checkThreadsShutdown(threads);
}
@Test
@Ignore
public void testSoakTestWithRandomData() throws IOException, InterruptedException {
Thread.sleep(100);
// for (int i = 30; i < 50; i++) {
System.out.print("SoakTesting ");
int max = 1000000;
for (int j = 1; j < max; j++) {
if (j % 100 == 0)
System.out.print(".");
Random rnd = new Random(System.currentTimeMillis());
final ChronicleMap<Integer, CharSequence> map = rnd.nextBoolean() ? map1 : map2;
if (rnd.nextBoolean()) {
map.put((int) rnd.nextInt(100), "test" + j);
} else {
map.remove((int) rnd.nextInt(100));
}
}
System.out.println("\nwaiting till equal");
waitTillUnchanged(1000);
System.out.println("time t=" + t);
Assert.assertEquals(new TreeMap(map1), new TreeMap(map2));
}
private void waitTillUnchanged(final int timeOutMs) throws InterruptedException {
Map map1UnChanged = new HashMap();
Map map2UnChanged = new HashMap();
int numberOfTimesTheSame = 0;
for (int t = 0; t < timeOutMs + 100; t++) {
if (map1.equals(map1UnChanged) && map2.equals(map2UnChanged)) {
numberOfTimesTheSame++;
} else {
numberOfTimesTheSame = 0;
map1UnChanged = new HashMap(map1);
map2UnChanged = new HashMap(map2);
}
Thread.sleep(50);
if (numberOfTimesTheSame == 100) {
break;
}
}
}
private void late(TimeProvider timeProvider) {
Mockito.when(timeProvider.currentTimeMillis()).thenReturn(System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(5));
}
}
src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java
+5 -3
View File
@@ -22,6 +22,8 @@ import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.UdpTransportConfig;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import java.io.Closeable;
import java.io.IOException;
@@ -75,8 +77,8 @@ public class UDPSocketReplicationTest {
System.gc();
}
/* @Test
@Ignore
@Test
@Ignore
public void testBufferOverflow() throws IOException, InterruptedException {
for (int i = 0; i < 1024; i++) {
@@ -84,7 +86,7 @@ public class UDPSocketReplicationTest {
map2.put(i * 2, "E");
System.out.println("" + map2);
}
}*/
}
}
src/test/java/net/openhft/chronicle/map/jrs166/map/ChronicleMapTest.java
-24
View File
@@ -23,7 +23,6 @@ import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.chronicle.map.jrs166.JSR166TestCase;
import org.junit.Test;
import java.io.File;
import java.io.IOException;
import java.util.*;
@@ -39,40 +38,17 @@ import static org.junit.Assert.*;
public class ChronicleMapTest extends JSR166TestCase {
private static File getPersistenceFile() {
String TMP = System.getProperty("java.io.tmpdir");
File file = new File(TMP + "/chm-test" + System.nanoTime());
file.deleteOnExit();
return file;
}
static ChronicleMap<Integer, CharSequence> newShmIntString(int size) throws IOException {
return ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(size).create();
}
static ChronicleMap<ArrayList, CharSequence> newShmListBoolean(int size) throws IOException {
return ChronicleMapBuilder.of(ArrayList.class, CharSequence.class)
.entries(size).create();
}
static ChronicleMap<ArrayList, CharSequence> newShmListBoolean() throws IOException {
return ChronicleMapBuilder.of(ArrayList.class, CharSequence.class).create();
}
static ChronicleMap<CharSequence, CharSequence> newStrStrMap(int size) throws IOException {
return ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
.entries(size).create();
}
static ChronicleMap<Integer, CharSequence> newShmIntString() throws IOException {
return ChronicleMapBuilder.of(Integer.class, CharSequence.class).create();
}
static ChronicleMap<BI, Boolean> newShmBiBoolean() throws IOException {
return ChronicleMapBuilder.of(BI.class, Boolean.class).create();
}
/**
* Returns a new map from Integers 1-5 to Strings "A"-"E".
*/