From 118e5c500657b6cbafca86ec927dfeb6f7d0d3a6 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 11:13:17 +0000 Subject: [PATCH 01/35] corrected javadoc --- .../java/net/openhft/chronicle/map/BytesMapEventListener.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/BytesMapEventListener.java b/src/main/java/net/openhft/chronicle/map/BytesMapEventListener.java index 5843ae8bc..2746dd9fd 100644 --- a/src/main/java/net/openhft/chronicle/map/BytesMapEventListener.java +++ b/src/main/java/net/openhft/chronicle/map/BytesMapEventListener.java @@ -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}). *
  • Methods must not update entry bytes, except in metadata area
    • *
  • 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()}).
    • + * should be itself aware of it's length (see {@link ChronicleMapBuilder#metaDataBytes(int)}). *
  • 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.
    • From 8ebcec3c0c698a522a4e5a451188671700294a2b Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 11:24:30 +0000 Subject: [PATCH 02/35] corrected javadoc --- .../openhft/chronicle/map/ChronicleMap.java | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java index fb5cd53e3..50a51e631 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java @@ -220,6 +220,7 @@ public interface ChronicleMap extends ConcurrentMap, 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,12 @@ public interface ChronicleMap extends ConcurrentMap, 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 key, or null - * if there was no mapping for key. (A null return can also indicate that the - * map previously associated null with key, if the implementation supports - * null 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 key, or null if there was no + * mapping for key. (A null return can also indicate that the map + * previously associated null with key, if the implementation + * supports null values.) * @see java.util.Map#put(java.lang.Object, java.lang.Object) * @see java.util.concurrent.Future */ @@ -279,8 +282,8 @@ public interface ChronicleMap extends ConcurrentMap, 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 +299,8 @@ public interface ChronicleMap extends ConcurrentMap, 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 From 5ab8428d000683f6a9db88a0e18557bf86e40bca Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 11:24:57 +0000 Subject: [PATCH 03/35] corrected javadoc and upgraded versions --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index f846e93d8..38b0e9097 100644 --- a/pom.xml +++ b/pom.xml @@ -48,13 +48,13 @@ net.openhft lang - 6.4.15-SNAPSHOT + 6.4.15 net.openhft affinity - 2.1.3 + 2.1.4 From 01982ba038a6b86bed9089d50150a542af18be45 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 11:57:17 +0000 Subject: [PATCH 04/35] imporved javadoc --- src/main/java/net/openhft/chronicle/map/ChronicleMap.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java index 50a51e631..e0cef2c72 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java @@ -243,6 +243,8 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { * mapping for key. (A null return can also indicate that the map * previously associated null with key, if the implementation * supports null values.) + * @return a future containing either, the previous value associated with key, or + * null if there was no mapping for key. * @see java.util.Map#put(java.lang.Object, java.lang.Object) * @see java.util.concurrent.Future */ From 03f3f58b91063a52e48035067ff08d6f034dc897 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 12:10:42 +0000 Subject: [PATCH 05/35] imporved javadoc --- .../openhft/chronicle/hash/ChronicleHashBuilder.java | 6 +++--- .../java/net/openhft/chronicle/hash/FindByName.java | 8 +++++--- .../hash/replication/TcpTransportAndNetworkConfig.java | 10 +++++++--- .../net/openhft/chronicle/set/ChronicleSetBuilder.java | 4 ++-- 4 files changed, 17 insertions(+), 11 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index 8cec3130c..7a12a592d 100755 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -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. Chronicle hash container configurations. * *

    {@code ChronicleHashBuilder} is mutable. Configuration methods mutate the builder and return @@ -131,7 +131,7 @@ public interface ChronicleHashBuilderThere are three major patterns of this configuration usage:

    1. 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.
    2. 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
    3. 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.
    4. 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 diff --git a/src/main/java/net/openhft/chronicle/hash/FindByName.java b/src/main/java/net/openhft/chronicle/hash/FindByName.java index dcfc27564..7f5ffbf3f 100644 --- a/src/main/java/net/openhft/chronicle/hash/FindByName.java +++ b/src/main/java/net/openhft/chronicle/hash/FindByName.java @@ -10,10 +10,12 @@ public interface FindByName { /** * @param name the name of the map or set - * @param + * @param 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 IllegalArgumentException if a map with this name can not be found + * @throws IOException if it not possible to create the map or set + * @throws java.util.concurrent.TimeoutException if the call times out + * @throws java.util.concurrent.TimeoutException if interrupted by another thread */ T from(String name) throws IllegalArgumentException, IOException, TimeoutException, InterruptedException; diff --git a/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java b/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java index ccd51fb49..f75804f6d 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java @@ -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; diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index 7430fe1c9..d669d3bec 100755 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -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. * - *

      {@code ChronicleMapOnHeapUpdatableBuilder} is mutable, see a note in {@link + *

      {@code ChronicleMapBuilder} is mutable, see a note in {@link * ChronicleHashBuilder} interface documentation. * * @param 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 implements ChronicleHashBuilder, ChronicleSetBuilder> { From 11c86a20571aa345e2889489dcaffbd3c4d464ef Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 12:20:43 +0000 Subject: [PATCH 06/35] imporved javadoc --- .../openhft/chronicle/hash/FindByName.java | 8 +- .../chronicle/map/ChronicleMapBuilderI.java | 142 +++++++++--------- .../OnHeapUpdatableChronicleMapBuilder.java | 8 +- 3 files changed, 83 insertions(+), 75 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/FindByName.java b/src/main/java/net/openhft/chronicle/hash/FindByName.java index 7f5ffbf3f..77fd8fcfa 100644 --- a/src/main/java/net/openhft/chronicle/hash/FindByName.java +++ b/src/main/java/net/openhft/chronicle/hash/FindByName.java @@ -12,10 +12,10 @@ public interface FindByName { * @param name the name of the map or set * @param 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 java.util.concurrent.TimeoutException if the call times out - * @throws java.util.concurrent.TimeoutException if interrupted by another thread + * @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 from(String name) throws IllegalArgumentException, IOException, TimeoutException, InterruptedException; diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java index ac81e54bb..ed4dd05d5 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java @@ -30,25 +30,25 @@ public interface ChronicleMapBuilderI extends Serializable { * {@inheritDoc} With respect to {@linkplain #entryAndValueAlignment(Alignment) alignment}. * *

      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 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. * - *

      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. + *

      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. * - *

      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). + *

      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). * *

      Default is {@link Alignment#OF_4_BYTES} for Byteable values. * @@ -58,11 +58,11 @@ public interface ChronicleMapBuilderI extends Serializable { ChronicleMapBuilderI 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}. * - *

      By default, the default value is not specified, default {@linkplain #prepareValueBytesOnAcquire - * prepare value bytes routine} is specified instead. + *

      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 extends Serializable { ChronicleMapBuilderI 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. * *

      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 extends Serializable { ChronicleMapBuilderI 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. * - *

      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. + *

      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. * - *

      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. + *

      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 extends Serializable { ChronicleMapBuilderI 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. * - *

      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. + *

      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. * - *

      If value size varies, method {@link #valueSize(int)} or {@link #entrySize(int)} should be called - * instead of this one. + *

      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 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 valueDeserializationFactory( @@ -174,8 +177,8 @@ public interface ChronicleMapBuilderI extends Serializable { /** * {@inheritDoc} * - *

      By default, default value provider is not specified, {@link #defaultValue(Object) default value} is - * specified instead. + *

      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 extends Serializable { @NotNull DefaultValueProvider defaultValueProvider); /** - * Configures the {@code BytesMarshaller} used to serialize/deserialize values to/from off-heap memory in - * maps, created by this builder. See the + * Configures the {@code BytesMarshaller} used to serialize/deserialize values to/from off-heap + * memory in maps, created by this builder. See the * section about serialization in ChronicleMap manual for more information. * * @param valueMarshaller the marshaller used to serialize values @@ -199,20 +202,21 @@ public interface ChronicleMapBuilderI extends Serializable { @NotNull BytesMarshaller valueMarshaller); /** - * Configures the marshallers, used to serialize/deserialize values to/from off-heap memory in maps, - * created by this builder. See the + * Configures the marshallers, used to serialize/deserialize values to/from off-heap memory in + * maps, created by this builder. See the * section about serialization in ChronicleMap manual for more information. * - *

      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)}. + *

      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)}. * - *

      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. + *

      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 → {@link net.openhft.lang.io.Bytes} writer (interop) + * @param valueWriter the new value object → {@link net.openhft.lang.io.Bytes} writer + * (interop) strategy + * @param valueReader the new {@link net.openhft.lang.io.Bytes} → value object reader * strategy - * @param valueReader the new {@link net.openhft.lang.io.Bytes} → value object reader strategy * @return this builder back * @see #valueMarshaller(BytesMarshaller) */ @@ -220,19 +224,21 @@ public interface ChronicleMapBuilderI extends Serializable { @NotNull BytesWriter valueWriter, @NotNull BytesReader 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. * *

      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 valueSizeMarshaller( @@ -295,13 +301,15 @@ public interface ChronicleMapBuilderI extends Serializable { ChronicleMapBuilderI 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 bootstrapOnlyLocalEntries(boolean bootstapOnlyLocalEntries); diff --git a/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java index 3b07010f0..5b7be09d9 100644 --- a/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java @@ -103,8 +103,8 @@ final class OnHeapUpdatableChronicleMapBuilder /** * 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.

      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.

      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.

      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 /** * 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.

      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.

      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 From a09cf94b3504f07bc4fe9bb41f345b29073e3905 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 12:26:19 +0000 Subject: [PATCH 07/35] imporved javadoc --- .../chronicle/map/MapEventListener.java | 58 ++++++++++--------- .../OnHeapUpdatableChronicleMapBuilder.java | 6 +- 2 files changed, 34 insertions(+), 30 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/MapEventListener.java b/src/main/java/net/openhft/chronicle/map/MapEventListener.java index 72b8e2679..5cf73f8b9 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEventListener.java +++ b/src/main/java/net/openhft/chronicle/map/MapEventListener.java @@ -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:

        - *
      • Map data backup / replication.
      • Logging, monitoring, debugging.
      + * Contains methods which are called when {@link ChronicleMap} key events occur. Typical use cases: + *
      • Map data backup / replication.
      • Logging, monitoring, debugging.
      * - *

      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. + *

      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. * *

      To configure {@code MapEventListener} for {@code ChronicleMap}, use {@link * AbstractChronicleMapBuilder#eventListener(MapEventListener)} method. * *

      See {@link #logging(String)} implementation. * - *

      All these calls are synchronous while a {@code ChronicleMap} lock is held so make them as quick as - * possible. + *

      All these calls are synchronous while a {@code ChronicleMap} lock is held so make them as + * quick as possible. * - *

      The {@code entry} passed to the methods is {@code Bytes} instance positioned at meta data area. See - * {@link ChronicleHashBuilder#metaDataBytes(int)} for more information. + *

      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 key type of the maps, trackable by this event listener * @param value type of the maps, trackable by this event listener @@ -51,15 +51,17 @@ public abstract class MapEventListener 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 the map key type - * @param the map value type + * @param the map key type + * @param the map value type + * @param prefix logging prefix * @return the logging event listener */ public static MapEventListener logging(String prefix) { + //noinspection unchecked return new LoggingMapEventListener(prefix); } @@ -90,40 +92,42 @@ public abstract class MapEventListener 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. * *

      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 diff --git a/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java index 5b7be09d9..eaed43fdf 100644 --- a/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java @@ -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.

      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. * @@ -149,7 +149,7 @@ final class OnHeapUpdatableChronicleMapBuilder * // this class hasn't implemented yet, just for example * .objectSerializer(new KryoObjectSerializer()) * .create();} - *

      This serializer is used to serialize both keys and values, if they both require this: loosely + *

      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 } /** - * {@inheritDoc}

      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 From 824a43c9786547834f0aaa38e1bc4b02121def9b Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 12:27:15 +0000 Subject: [PATCH 08/35] increased timeout --- .../chronicle/map/TCPSocketReplicationPostConnectionTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationPostConnectionTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationPostConnectionTest.java index b4c0b63df..5be332940 100755 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationPostConnectionTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationPostConnectionTest.java @@ -66,7 +66,7 @@ public class TCPSocketReplicationPostConnectionTest { new InetSocketAddress("localhost", port)); // allow time for the recompilation to resolve - waitTillEqual(500); + waitTillEqual(1500); assertEquals(map1, map2); assertTrue(!map1.isEmpty()); From 55ad73bb0c260121212acb3ebea8e9a828590488 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 12:37:27 +0000 Subject: [PATCH 09/35] increased timeout --- .../chronicle/map/TCPSocketReplicationPostConnectionTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationPostConnectionTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationPostConnectionTest.java index 5be332940..a389e53b2 100755 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationPostConnectionTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationPostConnectionTest.java @@ -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(1500); + waitTillEqual(15000); assertEquals(map1, map2); assertTrue(!map1.isEmpty()); From 73ec004fe63ea216f466e67815c17d8e34b296f9 Mon Sep 17 00:00:00 2001 From: Team City Date: Mon, 1 Dec 2014 12:41:16 +0000 Subject: [PATCH 10/35] [maven-release-plugin] prepare release chronicle-map-2.0.11b --- pom.xml | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/pom.xml b/pom.xml index 38b0e9097..a56f7183a 100644 --- a/pom.xml +++ b/pom.xml @@ -17,20 +17,18 @@ ~ limitations under the License. --> - + net.openhft java-parent-pom 1.1.2 - + 4.0.0 chronicle-map - 2.0.11b-SNAPSHOT + 2.0.11b OpenHFT/Chronicle-Map Chronicle-Map @@ -181,7 +179,7 @@ scm:git:git@github.com:OpenHFT/Chronicle-Map.git scm:git:git@github.com:OpenHFT/Chronicle-Map.git scm:git:git@github.com:OpenHFT/Chronicle-Map.git - master + chronicle-map-2.0.11b From 87270d064234b98a1562490c25b144727b12644b Mon Sep 17 00:00:00 2001 From: Team City Date: Mon, 1 Dec 2014 12:41:20 +0000 Subject: [PATCH 11/35] [maven-release-plugin] prepare for next development iteration --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index a56f7183a..4541cea2b 100644 --- a/pom.xml +++ b/pom.xml @@ -28,7 +28,7 @@ 4.0.0 chronicle-map - 2.0.11b + 2.0.12b-SNAPSHOT OpenHFT/Chronicle-Map Chronicle-Map @@ -179,7 +179,7 @@ scm:git:git@github.com:OpenHFT/Chronicle-Map.git scm:git:git@github.com:OpenHFT/Chronicle-Map.git scm:git:git@github.com:OpenHFT/Chronicle-Map.git - chronicle-map-2.0.11b + master From b3141fcb46bc75c40e62ddf2382f5e41f302e58c Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 1 Dec 2014 21:10:22 +0700 Subject: [PATCH 12/35] Fixed 2 bugs, causing EventListenerTestWithTCPSocketReplication#testRemoteNotifyRemove() test fail --- src/main/java/net/openhft/chronicle/map/ReadValue.java | 3 +++ .../java/net/openhft/chronicle/map/VanillaChronicleMap.java | 4 +++- .../map/EventListenerTestWithTCPSocketReplication.java | 1 - 3 files changed, 6 insertions(+), 2 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ReadValue.java b/src/main/java/net/openhft/chronicle/map/ReadValue.java index 8d925ce37..2b90e13d7 100644 --- a/src/main/java/net/openhft/chronicle/map/ReadValue.java +++ b/src/main/java/net/openhft/chronicle/map/ReadValue.java @@ -24,6 +24,9 @@ import org.jetbrains.annotations.NotNull; * @param result type, map value type or Bytes */ interface ReadValue { + /** + * Updates the entry position by valueSize. + */ RV readValue(@NotNull ThreadLocalCopies copies, Bytes entry, RV usingValue, long valueSize); RV readNull(); } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 4ba59a3e7..bb5de7251 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -384,7 +384,7 @@ class VanillaChronicleMap, } static abstract class ReadValueToBytes implements ReadValue { - private SizeMarshaller valueSizeMarshaller; + SizeMarshaller valueSizeMarshaller; abstract Bytes bytes(long valueSize); @@ -398,6 +398,7 @@ class VanillaChronicleMap, usingBytes = bytes(valueSize); valueSizeMarshaller.writeSize(usingBytes, valueSize); usingBytes.write(entry, entry.position(), valueSize); + entry.skip(valueSize); return usingBytes; } } @@ -434,6 +435,7 @@ class VanillaChronicleMap, @Override Bytes bytes(long valueSize) { + valueSize = valueSizeMarshaller.sizeEncodingSize(valueSize) + valueSize; if (lazyBytes != null) { if (lazyBytes.capacity() < valueSize) { DirectStore store = (DirectStore) lazyBytes.store(); diff --git a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java index 893b96a30..3d00cb137 100644 --- a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java +++ b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java @@ -180,7 +180,6 @@ public class EventListenerTestWithTCPSocketReplication { } @Test - @Ignore("TODO Failing test") public void testRemoteNotifyRemove() throws IOException, InterruptedException { // ---------- SERVER1 1 ---------- From 0439d0734e3e32d8ffdd73981440c22f3451d066 Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 1 Dec 2014 21:33:54 +0700 Subject: [PATCH 13/35] Test rename (TC problem) --- .../{TestBooleanValues.java => BooleanValuesTest.java} | 2 +- ...ava => EventListenerWithTCPSocketReplicationTest.java} | 2 +- .../openhft/chronicle/map/UDPSocketReplicationTest.java | 8 +++++--- 3 files changed, 7 insertions(+), 5 deletions(-) rename src/test/java/net/openhft/chronicle/map/{TestBooleanValues.java => BooleanValuesTest.java} (97%) rename src/test/java/net/openhft/chronicle/map/{EventListenerTestWithTCPSocketReplication.java => EventListenerWithTCPSocketReplicationTest.java} (99%) diff --git a/src/test/java/net/openhft/chronicle/map/TestBooleanValues.java b/src/test/java/net/openhft/chronicle/map/BooleanValuesTest.java similarity index 97% rename from src/test/java/net/openhft/chronicle/map/TestBooleanValues.java rename to src/test/java/net/openhft/chronicle/map/BooleanValuesTest.java index 20f10db07..2f8a71154 100644 --- a/src/test/java/net/openhft/chronicle/map/TestBooleanValues.java +++ b/src/test/java/net/openhft/chronicle/map/BooleanValuesTest.java @@ -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 diff --git a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java b/src/test/java/net/openhft/chronicle/map/EventListenerWithTCPSocketReplicationTest.java similarity index 99% rename from src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java rename to src/test/java/net/openhft/chronicle/map/EventListenerWithTCPSocketReplicationTest.java index 3d00cb137..8eb4afd4d 100644 --- a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java +++ b/src/test/java/net/openhft/chronicle/map/EventListenerWithTCPSocketReplicationTest.java @@ -35,7 +35,7 @@ import java.util.concurrent.atomic.AtomicReference; * * @author Rob Austin. */ -public class EventListenerTestWithTCPSocketReplication { +public class EventListenerWithTCPSocketReplicationTest { ChronicleMap map1; ChronicleMap map2; diff --git a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java index 85d572a40..2ef6178ef 100644 --- a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java @@ -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); } - }*/ + } } From 427fa5064f31b7945dfdbe710b0f524996eb109c Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 1 Dec 2014 21:41:57 +0700 Subject: [PATCH 14/35] HCOLL-165 seems that now ConstantSizeBySampleTest#testUnexpectedlyLongConstantSerializableValues() works --- .../net/openhft/chronicle/map/ConstantSizeBySampleTest.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/ConstantSizeBySampleTest.java b/src/test/java/net/openhft/chronicle/map/ConstantSizeBySampleTest.java index 0e5481e40..a570ecd07 100644 --- a/src/test/java/net/openhft/chronicle/map/ConstantSizeBySampleTest.java +++ b/src/test/java/net/openhft/chronicle/map/ConstantSizeBySampleTest.java @@ -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 map = ChronicleMapBuilder.of(Long.class, SerializableData.class) From b774cc3e9a4ff86a8489754019f515582af811f2 Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 1 Dec 2014 22:07:31 +0700 Subject: [PATCH 15/35] Replair Javadoc links to Lang --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 4541cea2b..28d3ca597 100644 --- a/pom.xml +++ b/pom.xml @@ -165,7 +165,7 @@ maven-javadoc-plugin - http://openhft.github.io/Chronicle-Map/apidocs/ + http://openhft.github.io/Java-Lang/apidocs/ net.openhft.chronicle.hash.serialization.internal From f0bb095d65beb57bbf230917a3fd8128532e1393 Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 1 Dec 2014 23:21:58 +0700 Subject: [PATCH 16/35] HCOLL-222 clarified error message if trying to put the unitialized byteable key/value, that might be quite common mistake --- .../hash/serialization/internal/ByteableMarshaller.java | 9 ++++++++- .../java/net/openhft/chronicle/map/ChronicleMapTest.java | 3 +-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java index 2e7f415ce..7e7dd742c 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java @@ -103,7 +103,14 @@ public class ByteableMarshaller @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 diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index 54906f545..60f897a35 100755 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -1398,8 +1398,7 @@ public class ChronicleMapTest { } } - @Ignore("JIRA raised for failing test HCOLL-222") - @Test + @Test(expected = NullPointerException.class) public void testPutLongValue() throws IOException { final ChronicleMapBuilder builder = ChronicleMapBuilder .of(CharSequence.class, LongValue.class) From cf1022418b3857e97b93fee7d7e3aef44a922022 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 18:03:50 +0000 Subject: [PATCH 17/35] added test net.openhft.chronicle.map.TcpReplicationSoakTest.testSoakTestWithRandomData --- .../chronicle/map/tcpReplicationSoakTest.java | 137 ++++++++++++++++++ 1 file changed, 137 insertions(+) create mode 100644 src/test/java/net/openhft/chronicle/map/tcpReplicationSoakTest.java diff --git a/src/test/java/net/openhft/chronicle/map/tcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/tcpReplicationSoakTest.java new file mode 100644 index 000000000..d256d0d5f --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/tcpReplicationSoakTest.java @@ -0,0 +1,137 @@ +package net.openhft.chronicle.map; + +import junit.framework.Assert; +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.After; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.Closeable; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +/** + * @author Rob Austin. + */ +public class TcpReplicationSoakTest { + + + private ChronicleMap map1; + private ChronicleMap 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) + .heartBeatInterval(1L, TimeUnit.SECONDS) + .autoReconnectedUponDroppedConnection(true); + + + map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(Builder.SIZE) + .replication((byte) 1, tcpConfig1) + .create(); + } + { + final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of(s_port + 1) + .heartBeatInterval(1L, TimeUnit.SECONDS) + .autoReconnectedUponDroppedConnection(true); + + map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(Builder.SIZE) + .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 threads; + + @Before + public void sampleThreads() { + threads = Thread.getAllStackTraces().keySet(); + } + + @After + public void checkThreadsShutdown() { + StatelessClientTest.checkThreadsShutdown(threads); + } + + + // TODO test this with larger sizes. + @Test(timeout = 20000) + @Ignore("Doesn't work, maps need to check equality.") + public void testSoakTestWithRandomData() throws IOException, InterruptedException { + + Thread.sleep(100); + final long start = System.currentTimeMillis(); + 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 map = select > 0 ? map1 : map2; + + if (rnd.nextBoolean()) { + map.put(rnd.nextInt(Builder.SIZE), "test" + j); + } else { + map.remove(rnd.nextInt(Builder.SIZE)); + } + } + } + + System.out.println("\nwaiting till equal"); + + waitTillEqual(1000); + + Assert.assertEquals(map1, map2); + + } + + private void waitTillEqual(final int timeOutMs) throws InterruptedException { + int numberOfTimesTheSame = 0; + for (int t = 0; t < timeOutMs + 100; t++) { + if (map1.equals(map2)) { + numberOfTimesTheSame++; + Thread.sleep(1); + if (numberOfTimesTheSame == 100) + break; + } else numberOfTimesTheSame = 0; + Thread.sleep(1); + } + } +} From 5239cec01670534f923695feb740c8b1d6299e74 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 18:18:42 +0000 Subject: [PATCH 18/35] Rename tcpReplicationSoakTest.java to TcpReplicationSoakTest.java rename --- .../{tcpReplicationSoakTest.java => TcpReplicationSoakTest.java} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename src/test/java/net/openhft/chronicle/map/{tcpReplicationSoakTest.java => TcpReplicationSoakTest.java} (100%) diff --git a/src/test/java/net/openhft/chronicle/map/tcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java similarity index 100% rename from src/test/java/net/openhft/chronicle/map/tcpReplicationSoakTest.java rename to src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java From 397549ac52584d6ee03186168b5d33d01fe2174d Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 1 Dec 2014 19:49:19 +0000 Subject: [PATCH 19/35] categorised @ignore with JIRA's and fixed test cases and List Marshaller --- pom.xml | 8 ++- .../chronicle/map/VanillaChronicleMap.java | 26 ++++++--- .../chronicle/map/CHMUseCasesTest.java | 55 +++++++++++++------ .../map/jrs166/map/ChronicleMapTest.java | 24 -------- .../chronicle/map/tcpReplicationSoakTest.java | 4 +- 5 files changed, 63 insertions(+), 54 deletions(-) diff --git a/pom.xml b/pom.xml index 28d3ca597..55df114de 100644 --- a/pom.xml +++ b/pom.xml @@ -17,13 +17,15 @@ ~ limitations under the License. --> - + net.openhft java-parent-pom 1.1.2 - + 4.0.0 @@ -46,7 +48,7 @@ net.openhft lang - 6.4.15 + 6.4.16-SNAPSHOT diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index bb5de7251..563e38bc5 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -316,18 +316,18 @@ class VanillaChronicleMap, } 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) { @@ -1176,6 +1176,10 @@ class VanillaChronicleMap, 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); } @@ -1187,6 +1191,9 @@ class VanillaChronicleMap, @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); } @@ -1597,6 +1604,7 @@ class VanillaChronicleMap, } long startWriteLock = 0; + @Override public final WriteLocked writeLock( @Nullable SegmentState segmentState) { diff --git a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java index 831575868..6c98eb444 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java +++ b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java @@ -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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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.emptyList()); - map.put("2", Arrays.asList("one")); + map.put("2", Arrays.asList("two-A")); List list1 = new ArrayList<>(); try (WriteContext wc = map.acquireUsingLocked("1", list1)) { - list1.add("two"); - assertEquals(Arrays.asList("two"), list1); + assertEquals(Collections.EMPTY_LIST, list1); } + List 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> map = ChronicleMapBuilder .of(String.class, (Class>) (Class) Set.class) diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/ChronicleMapTest.java index 3e30638d2..5024f490b 100644 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/ChronicleMapTest.java @@ -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 newShmIntString(int size) throws IOException { return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(size).create(); } - static ChronicleMap newShmListBoolean(int size) throws IOException { - return ChronicleMapBuilder.of(ArrayList.class, CharSequence.class) - .entries(size).create(); - } - - static ChronicleMap newShmListBoolean() throws IOException { - return ChronicleMapBuilder.of(ArrayList.class, CharSequence.class).create(); - } - static ChronicleMap newStrStrMap(int size) throws IOException { return ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) .entries(size).create(); } - static ChronicleMap newShmIntString() throws IOException { - return ChronicleMapBuilder.of(Integer.class, CharSequence.class).create(); - } - - static ChronicleMap newShmBiBoolean() throws IOException { - return ChronicleMapBuilder.of(BI.class, Boolean.class).create(); - } - /** * Returns a new map from Integers 1-5 to Strings "A"-"E". */ diff --git a/src/test/java/net/openhft/chronicle/map/tcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/tcpReplicationSoakTest.java index d256d0d5f..db18f47a4 100644 --- a/src/test/java/net/openhft/chronicle/map/tcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/tcpReplicationSoakTest.java @@ -95,8 +95,8 @@ public class TcpReplicationSoakTest { @Ignore("Doesn't work, maps need to check equality.") public void testSoakTestWithRandomData() throws IOException, InterruptedException { - Thread.sleep(100); - final long start = System.currentTimeMillis(); + + System.out.print("SoakTesting "); for (int j = 1; j < 2 * Builder.SIZE; j++) { if (j % 100 == 0) From a5631953d63d5665a215bcf6eb0910efbd630064 Mon Sep 17 00:00:00 2001 From: leventov Date: Tue, 2 Dec 2014 14:49:20 +0700 Subject: [PATCH 20/35] HCOLL-221 enabled testAcquireFromMultipleThreads() test, because failing ONLY within IntelliJ, from Maven build it succeeds on all machines. Seems there is no bugs. --- src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index 60f897a35..7c8bcd560 100755 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -523,7 +523,6 @@ public class ChronicleMapTest { map.close(); } - @Ignore("JIRA raised HCOLL-221") @Test public void testAcquireFromMultipleThreads() throws Exception { int entries = 1000 * 1000; @@ -551,7 +550,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(); } From 622a62f206ebe5124d80585d6cf5ec1219d9aa82 Mon Sep 17 00:00:00 2001 From: leventov Date: Tue, 2 Dec 2014 14:56:46 +0700 Subject: [PATCH 21/35] Don't free blocks of removed entries in ReplicatedChronicleMap --- .../openhft/chronicle/map/VanillaChronicleMap.java | 13 +++---------- .../chronicle/map/TcpReplicationSoakTest.java | 11 ++--------- 2 files changed, 5 insertions(+), 19 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 563e38bc5..1dee27fc5 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -2031,10 +2031,6 @@ class VanillaChronicleMap, 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; @@ -2165,12 +2161,12 @@ class VanillaChronicleMap, // 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) @@ -2383,9 +2379,6 @@ class VanillaChronicleMap, // 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); @@ -2595,11 +2588,11 @@ class VanillaChronicleMap, 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); diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index db18f47a4..bc538fdf5 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -1,15 +1,11 @@ package net.openhft.chronicle.map; -import junit.framework.Assert; 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.After; -import org.junit.Before; -import org.junit.Ignore; -import org.junit.Test; +import org.junit.*; import java.io.Closeable; import java.io.IOException; @@ -92,11 +88,8 @@ public class TcpReplicationSoakTest { // TODO test this with larger sizes. @Test(timeout = 20000) - @Ignore("Doesn't work, maps need to check equality.") public void testSoakTestWithRandomData() throws IOException, InterruptedException { - - System.out.print("SoakTesting "); for (int j = 1; j < 2 * Builder.SIZE; j++) { if (j % 100 == 0) @@ -116,7 +109,7 @@ public class TcpReplicationSoakTest { System.out.println("\nwaiting till equal"); - waitTillEqual(1000); + waitTillEqual(10000); Assert.assertEquals(map1, map2); From c4417ddbe49e0fcc8e6ccaa6456265546a39662b Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 2 Dec 2014 20:08:26 +0000 Subject: [PATCH 22/35] adding net.openhft.chronicle.map.NoTcpReplicationSoakTest --- .../chronicle/map/ReplicatedChronicleMap.java | 150 +++++++----- .../openhft/chronicle/map/TcpReplicator.java | 19 +- .../map/NoTcpReplicationSoakTest.java | 230 ++++++++++++++++++ .../map/TCPSocketReplicationTest.java | 45 +++- .../chronicle/map/TcpReplicationSoakTest.java | 64 +++-- .../map/TcpTimeBasedReplicationSoakTest.java | 166 +++++++++++++ 6 files changed, 579 insertions(+), 95 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java create mode 100644 src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 9b517eb0c..0ce3d5567 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -48,36 +48,39 @@ import static net.openhft.lang.MemoryUnit.*; import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND; /** - *

      A Replicating Multi Master HashMap

      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.

      Reconciliation

      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.

      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.

      + *

      A Replicating Multi Master HashMap

      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.

      + *

      Reconciliation

      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.

      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.

      * * @param the entries key type * @param the entries value type @@ -207,7 +210,8 @@ final class ReplicatedChronicleMap>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, RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> @@ -380,8 +412,8 @@ final class ReplicatedChronicleMap, RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> @@ -1133,8 +1162,8 @@ final class ReplicatedChronicleMapOnce 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.

      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.

      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.

      + *

      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. + *

      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.

      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.

      * * @author Rob Austin. */ @@ -1344,7 +1374,8 @@ final class ReplicatedChronicleMap 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 +662,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 +1022,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 +1032,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 +1199,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 +1251,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 +1273,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() { diff --git a/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java new file mode 100644 index 000000000..17a5c256b --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java @@ -0,0 +1,230 @@ +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 map1; + private ReplicatedChronicleMap 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 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() { + task.decrementAndGet(); + map2.put0(key, value, true, (byte) 1, t); + } + }); + + } else { + map2.put(key, value); + + task.incrementAndGet(); + executorService.submit(new Runnable() { + @Override + public void run() { + task.decrementAndGet(); + map1.put0(key, value, true, (byte) 2, t); + } + }); + } + + } else { + + if (rnd.nextBoolean()) { + map1.remove(key); + + task.incrementAndGet(); + executorService.submit(new Runnable() { + @Override + public void run() { + task.decrementAndGet(); + map2.remoteRemove(key, (byte) 1, time); + } + }); + + } else { + map2.remove(key); + + task.incrementAndGet(); + executorService.submit(new Runnable() { + @Override + public void run() { + task.decrementAndGet(); + map1.remoteRemove(key, (byte) 2, time); + } + }); + } + } + } + + 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(); + } + } + }); + } + } + + + 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(500); + if (numberOfTimesTheSame == 100) { + System.out.println("same"); + break; + } + + } + Thread.sleep(1); + } + } + +} + diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java index fff7574f5..32df4b0bd 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java @@ -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 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); } } diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index bc538fdf5..825f7b2da 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -11,9 +11,7 @@ import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; -import java.util.Random; -import java.util.Set; -import java.util.concurrent.TimeUnit; +import java.util.*; /** * @author Rob Austin. @@ -35,25 +33,26 @@ public class TcpReplicationSoakTest { { final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port, - endpoint) - .heartBeatInterval(1L, TimeUnit.SECONDS) - .autoReconnectedUponDroppedConnection(true); + endpoint).autoReconnectedUponDroppedConnection(true).name(" map1"); map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(Builder.SIZE) + .entries(Builder.SIZE + Builder.SIZE) + .actualSegments(1) + .name("map1") .replication((byte) 1, tcpConfig1) .create(); } { - final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of(s_port + 1) - .heartBeatInterval(1L, TimeUnit.SECONDS) - .autoReconnectedUponDroppedConnection(true); + final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of + (s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2"); map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(Builder.SIZE) + .entries(Builder.SIZE + Builder.SIZE) + .name("map2") .replication((byte) 2, tcpConfig2) .create(); + } s_port += 2; } @@ -86,8 +85,8 @@ public class TcpReplicationSoakTest { } - // TODO test this with larger sizes. - @Test(timeout = 20000) + @Test + @Ignore public void testSoakTestWithRandomData() throws IOException, InterruptedException { System.out.print("SoakTesting "); @@ -99,32 +98,47 @@ public class TcpReplicationSoakTest { final int select = rnd.nextInt(2); final ChronicleMap map = select > 0 ? map1 : map2; - if (rnd.nextBoolean()) { - map.put(rnd.nextInt(Builder.SIZE), "test" + j); - } else { - map.remove(rnd.nextInt(Builder.SIZE)); - } + 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); + waitTillEqual(10000); - Assert.assertEquals(map1, map2); + 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)) { - numberOfTimesTheSame++; - Thread.sleep(1); - if (numberOfTimesTheSame == 100) + if (map1.equals(map1UnChanged) && map2.equals(map2UnChanged)) { + numberOfTimesTheSame++; + } else { + numberOfTimesTheSame = 0; + map1UnChanged = new HashMap(map1); + map2UnChanged = new HashMap(map2); + } + Thread.sleep(500); + if (numberOfTimesTheSame == 100) { + System.out.println("same"); break; - } else numberOfTimesTheSame = 0; + } + + } Thread.sleep(1); } } } + diff --git a/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java new file mode 100644 index 000000000..d2e5b5720 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java @@ -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 map1; + private ChronicleMap 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 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 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)); + } +} + From 1b06733c6733cf9cdbccfa47ae79f58bfedce378 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 09:21:37 +0000 Subject: [PATCH 23/35] the decrements were in the wrong place --- .../map/NoTcpReplicationSoakTest.java | 38 ++++--------------- 1 file changed, 8 insertions(+), 30 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java index 17a5c256b..23579cb93 100644 --- a/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java @@ -44,7 +44,6 @@ public class NoTcpReplicationSoakTest { map1 = (ReplicatedChronicleMap) ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .name("map1") .replication((byte) 1) .timeProvider(timeProvider) @@ -53,7 +52,6 @@ public class NoTcpReplicationSoakTest { { map2 = (ReplicatedChronicleMap) ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .name("map2") .replication((byte) 2) .timeProvider(timeProvider) @@ -125,8 +123,9 @@ public class NoTcpReplicationSoakTest { executorService.submit(new Runnable() { @Override public void run() { - task.decrementAndGet(); + map2.put0(key, value, true, (byte) 1, t); + task.decrementAndGet(); } }); @@ -137,8 +136,9 @@ public class NoTcpReplicationSoakTest { executorService.submit(new Runnable() { @Override public void run() { - task.decrementAndGet(); + map1.put0(key, value, true, (byte) 2, t); + task.decrementAndGet(); } }); } @@ -152,8 +152,9 @@ public class NoTcpReplicationSoakTest { executorService.submit(new Runnable() { @Override public void run() { - task.decrementAndGet(); + map2.remoteRemove(key, (byte) 1, time); + task.decrementAndGet(); } }); @@ -164,8 +165,9 @@ public class NoTcpReplicationSoakTest { executorService.submit(new Runnable() { @Override public void run() { - task.decrementAndGet(); + map1.remoteRemove(key, (byte) 2, time); + task.decrementAndGet(); } }); } @@ -200,31 +202,7 @@ public class NoTcpReplicationSoakTest { } - 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(500); - if (numberOfTimesTheSame == 100) { - System.out.println("same"); - break; - } - - } - Thread.sleep(1); - } - } } From 5f1d94ead806ccb7a604544f4612733a86cfe04c Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 09:22:51 +0000 Subject: [PATCH 24/35] remove was using the wrong time --- .../net/openhft/chronicle/map/NoTcpReplicationSoakTest.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java index 23579cb93..b1003d1de 100644 --- a/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java @@ -112,7 +112,7 @@ public class NoTcpReplicationSoakTest { final int key = rnd.nextInt(1000); // add random delay - addRandomDelay(executorService, rnd); + addRandomDelay(executorService, rnd); if (rnd.nextBoolean()) { final CharSequence value = "test" + j; @@ -153,7 +153,7 @@ public class NoTcpReplicationSoakTest { @Override public void run() { - map2.remoteRemove(key, (byte) 1, time); + map2.remoteRemove(key, (byte) 1, t); task.decrementAndGet(); } }); @@ -166,7 +166,7 @@ public class NoTcpReplicationSoakTest { @Override public void run() { - map1.remoteRemove(key, (byte) 2, time); + map1.remoteRemove(key, (byte) 2, t); task.decrementAndGet(); } }); From 81082aab62ee49ea8317e00a53c0e22b398d1eff Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 09:29:56 +0000 Subject: [PATCH 25/35] Added back in the @Ignore and JIRA - this has just failed on TC so I defiantly think there is a genuine bug MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ChronicleMapTest.testAcquireFromMultipleThreads Current failure: #816 rob.austin (2) 03 Dec 14 09:23 First failure: #815 rob.austin (1) 02 Dec 14 20:09 java.lang.AssertionError: expected:<10000> but was:<9994> at org.junit.Assert.fail(Assert.java:88) at org.junit.Assert.failNotEquals(Assert.java:743) at org.junit.Assert.assertEquals(Assert.java:118) at org.junit.Assert.assertEquals(Assert.java:555) at org.junit.Assert.assertEquals(Assert.java:542) at net.openhft.chronicle.map.ChronicleMapTest.testAcquireFromMultipleThreads(ChronicleMapTest.java:553) at net.openhft.chronicle.map.ChronicleMapTest.testAcquireFromMultipleThreads(ChronicleMapTest.java:529) « Hide stacktrace see http://teamcity.higherfrequencytrading.com:8111/viewLog.html?buildId=3763&tab=buildResultsDiv&buildTypeId=OpenHFT_ChronicleMap_Snapshot --- src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index 7c8bcd560..ea9a9ba7d 100755 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -523,6 +523,7 @@ public class ChronicleMapTest { map.close(); } + @Ignore("JIRA raised HCOLL-221") @Test public void testAcquireFromMultipleThreads() throws Exception { int entries = 1000 * 1000; From 2541d5acfd2c10a0225054b6263796479a417e7d Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 10:06:26 +0000 Subject: [PATCH 26/35] HCOLL-229 fix to -stateless client : Terminated thread causes next thread to read the other threads message --- .../chronicle/map/StatelessChronicleMap.java | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index b73bac2d4..afc71c411 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -938,9 +938,6 @@ class StatelessChronicleMap implements ChronicleMap, 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 +960,9 @@ class StatelessChronicleMap implements ChronicleMap, 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(); @@ -982,9 +982,15 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon 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); + 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(); From 2e2798f3180c6b5d7c24d950b463d227e3494265 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 10:07:42 +0000 Subject: [PATCH 27/35] fixed typo on comment --- .../java/net/openhft/chronicle/map/StatelessChronicleMap.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index afc71c411..e5f886d8b 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -976,7 +976,7 @@ class StatelessChronicleMap implements ChronicleMap, 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(); From ec6e99224d90e09bb00e740957dab6aa1b2be6dd Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 3 Dec 2014 20:26:52 +0700 Subject: [PATCH 28/35] Remote udpates remove positions from modifications iterators (previously was not touching modification iterators; fixed a small bug in ReplicatedCHronicleMap.removeWithoutUnlock(); fixed soak tests with random data --- .../chronicle/map/ReplicatedChronicleMap.java | 16 ++++++-- .../openhft/chronicle/map/TcpReplicator.java | 3 ++ .../chronicle/map/VanillaChronicleMap.java | 38 +++++++++++++++---- .../map/TCPSocketReplicationIntValueTest.java | 3 +- .../chronicle/map/TcpReplicationSoakTest.java | 10 +++-- 5 files changed, 53 insertions(+), 17 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 0ce3d5567..eebe5cc7a 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -387,6 +387,11 @@ final class ReplicatedChronicleMap.Segment segment, long pos) { + onRelocation(segment, pos); // erase modification bits + } + @Override void onRemove(VanillaChronicleMap.Segment segment, long pos) { for (long next = modIterSet.nextSetBit(0L); next > 0L; @@ -399,6 +404,11 @@ final class ReplicatedChronicleMap.Segment segment, long pos) { + onRelocation(segment, pos); // erase modification bits + } + @Override void onRelocation(VanillaChronicleMap.Segment segment, long pos) { for (long next = modIterSet.nextSetBit(0L); next > 0L; @@ -931,8 +941,7 @@ final class ReplicatedChronicleMap, // 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 } @@ -1940,8 +1948,7 @@ class VanillaChronicleMap, 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) { @@ -1952,6 +1959,14 @@ class VanillaChronicleMap, 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 @@ -2169,8 +2184,7 @@ class VanillaChronicleMap, decrementSize(); // remove callbacks - if (!remote) - onRemove(this, pos); + onRemoveMaybeRemote(pos, remote); if (bytesEventListener != null) bytesEventListener.onRemove(entry, 0L, metaDataBytes, valueSizePos); if (eventListener != null) { @@ -2183,6 +2197,14 @@ class VanillaChronicleMap, return booleanResult ? Boolean.TRUE : removedValue; } + final void onRemoveMaybeRemote(long pos, boolean remote) { + if (remote) { + onRemoteRemove(this, pos); + } else { + onRemove(this, pos); + } + } + private > boolean containsKey(ThreadLocalCopies copies, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, long hash2) { @@ -2337,15 +2359,15 @@ class VanillaChronicleMap, MetaBytesWriter 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) { diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java index 6a907c631..fc45a19ed 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java @@ -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 "); diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index 825f7b2da..14dd1dcfe 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -12,6 +12,7 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.*; +import java.util.concurrent.TimeUnit; /** * @author Rob Austin. @@ -33,7 +34,8 @@ public class TcpReplicationSoakTest { { final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port, - endpoint).autoReconnectedUponDroppedConnection(true).name(" map1"); + endpoint).autoReconnectedUponDroppedConnection(true).name(" map1") + .heartBeatInterval(1, TimeUnit.SECONDS); map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) @@ -45,7 +47,8 @@ public class TcpReplicationSoakTest { } { final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of - (s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2"); + (s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2") + .heartBeatInterval(1, TimeUnit.SECONDS); map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(Builder.SIZE + Builder.SIZE) @@ -86,7 +89,6 @@ public class TcpReplicationSoakTest { @Test - @Ignore public void testSoakTestWithRandomData() throws IOException, InterruptedException { System.out.print("SoakTesting "); @@ -130,7 +132,7 @@ public class TcpReplicationSoakTest { map1UnChanged = new HashMap(map1); map2UnChanged = new HashMap(map2); } - Thread.sleep(500); + Thread.sleep(1); if (numberOfTimesTheSame == 100) { System.out.println("same"); break; From 1cebaf8ca2b2f05ab0d600ce364a8ab985832c53 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 13:41:24 +0000 Subject: [PATCH 29/35] added BGChronicleTest --- src/test/java/eg/BGChronicleTest.java | 174 ++++++++++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 src/test/java/eg/BGChronicleTest.java diff --git a/src/test/java/eg/BGChronicleTest.java b/src/test/java/eg/BGChronicleTest.java new file mode 100644 index 000000000..981f25c8a --- /dev/null +++ b/src/test/java/eg/BGChronicleTest.java @@ -0,0 +1,174 @@ +package eg; + +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.map.ChronicleMap; +import net.openhft.chronicle.map.ChronicleMapBuilder; +import org.junit.Test; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.HashMap; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; + +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 startChronicleMapServer(int port) throws InterruptedException, IOException { + + + TcpTransportAndNetworkConfig map2Config = 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, map2Config) + .create(); + } + + /** + * over loop-back on my mac I get around 50,000 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 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); + + } + } + + private void logMessagesPerSecond(double numberOfMessages, long start) { + double messagesPerSecond = numberOfMessages / (double) TimeUnit.NANOSECONDS + .toSeconds(System.nanoTime() - start); + + System.out.println("messages per seconds " + messagesPerSecond); + } + + + @Test(timeout = 10000) + public void testBufferOverFlowPutAllAndEntrySet() throws IOException, InterruptedException { + int SIZE = 10; + int port = 9050; + try (ChronicleMap serverMap = ChronicleMapBuilder + .of(Integer.class, CharSequence.class) + .entries(SIZE) + .replication((byte) 2, TcpTransportAndNetworkConfig.of(port)) + .create()) { + try (ChronicleMap statelessMap = ChronicleMapBuilder + .of(Integer.class, CharSequence.class) + .statelessClient(new InetSocketAddress("localhost", port)) + .create()) { + Map payload = new HashMap(); + + for (int i = 0; i < SIZE; i++) { + payload.put(i, "some value=" + i); + } + + statelessMap.putAll(payload); + + Set> entries = statelessMap.entrySet(); + + Map.Entry next = entries.iterator().next(); + assertEquals("some value=" + next.getKey(), next.getValue()); + + assertEquals(SIZE, entries.size()); + } + } + } + + + /** + * for loop back args = [] + * + * for servereg.BGChronicleTest server + * + * @param args + * @throws IOException + * @throws InterruptedException + */ + public static void main(String... args) throws IOException, InterruptedException { + + final BGChronicleTest o = new BGChronicleTest(); + ChronicleMap serverMap = null; + try { + + if (args.length == 0 || (args.length > 1 && args[0].equalsIgnoreCase("server"))) { + int port = args.length > 0 && args[1] != null ? Integer.valueOf(args[1]) : DEFAULT_PORT; + serverMap = o.startChronicleMapServer(port); + System.out.println("server started."); + } + + if (args.length == 0 || (args.length > 1 && args[0].equalsIgnoreCase("client"))) { + String host = args.length > 0 && args[1] != null ? args[1] : DEFAULT_HOST; + int port = args.length > 1 && args[2] != null ? Integer.valueOf(args[2]) : + DEFAULT_PORT; + o.startChronicleMapClient(host, port, MAX_NUMBER_OF_EATERIES); + } + + } finally { + if (serverMap != null) + serverMap.close(); + } + + } + + public void test() throws IOException, InterruptedException { + main(); + } + +} From 573aa4e71e3d24568de78280fccb4ae8593fbe27 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 14:29:10 +0000 Subject: [PATCH 30/35] update performance test --- src/test/java/eg/BGChronicleTest.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/test/java/eg/BGChronicleTest.java b/src/test/java/eg/BGChronicleTest.java index 981f25c8a..a8a643e50 100644 --- a/src/test/java/eg/BGChronicleTest.java +++ b/src/test/java/eg/BGChronicleTest.java @@ -149,15 +149,20 @@ public class BGChronicleTest { if (args.length == 0 || (args.length > 1 && args[0].equalsIgnoreCase("server"))) { int port = args.length > 0 && args[1] != null ? Integer.valueOf(args[1]) : DEFAULT_PORT; + System.out.println("server starting...."); serverMap = o.startChronicleMapServer(port); System.out.println("server started."); - } + } else + System.out.println("server if OFF"); if (args.length == 0 || (args.length > 1 && args[0].equalsIgnoreCase("client"))) { String host = args.length > 0 && args[1] != null ? args[1] : DEFAULT_HOST; int port = args.length > 1 && args[2] != null ? Integer.valueOf(args[2]) : DEFAULT_PORT; + System.out.println("client starting...."); o.startChronicleMapClient(host, port, MAX_NUMBER_OF_EATERIES); + } else { + System.out.println("client if OFF"); } } finally { From d6b8ebd844d6e7fc886a0b3c7e92b9a19f72f8cd Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 14:33:38 +0000 Subject: [PATCH 31/35] updated test --- src/test/java/eg/BGChronicleTest.java | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/src/test/java/eg/BGChronicleTest.java b/src/test/java/eg/BGChronicleTest.java index a8a643e50..23d018844 100644 --- a/src/test/java/eg/BGChronicleTest.java +++ b/src/test/java/eg/BGChronicleTest.java @@ -147,19 +147,20 @@ public class BGChronicleTest { ChronicleMap serverMap = null; try { - if (args.length == 0 || (args.length > 1 && args[0].equalsIgnoreCase("server"))) { - int port = args.length > 0 && args[1] != null ? Integer.valueOf(args[1]) : DEFAULT_PORT; - System.out.println("server starting...."); + if (args.length == 0 || (args.length > 0 && args[0].equalsIgnoreCase("server"))) { + int port = args.length > 1 && args[1] != null ? Integer.valueOf(args[1]) : + DEFAULT_PORT; + serverMap = o.startChronicleMapServer(port); - System.out.println("server started."); + System.out.println("server ON"); } else System.out.println("server if OFF"); if (args.length == 0 || (args.length > 1 && args[0].equalsIgnoreCase("client"))) { - String host = args.length > 0 && args[1] != null ? args[1] : DEFAULT_HOST; - int port = args.length > 1 && args[2] != null ? Integer.valueOf(args[2]) : + String host = args.length > 1 && args[1] != null ? args[1] : DEFAULT_HOST; + int port = args.length > 2 && args[2] != null ? Integer.valueOf(args[2]) : DEFAULT_PORT; - System.out.println("client starting...."); + System.out.println("client ON...."); o.startChronicleMapClient(host, port, MAX_NUMBER_OF_EATERIES); } else { System.out.println("client if OFF"); From 40c791671eec914335adf01d323edb463f566eb7 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 14:42:27 +0000 Subject: [PATCH 32/35] tweeked the test --- src/test/java/eg/BGChronicleTest.java | 41 ++++----------------------- 1 file changed, 5 insertions(+), 36 deletions(-) diff --git a/src/test/java/eg/BGChronicleTest.java b/src/test/java/eg/BGChronicleTest.java index 23d018844..d7bca7df5 100644 --- a/src/test/java/eg/BGChronicleTest.java +++ b/src/test/java/eg/BGChronicleTest.java @@ -3,19 +3,14 @@ package eg; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; -import org.junit.Test; import java.io.IOException; import java.net.InetSocketAddress; -import java.util.HashMap; -import java.util.Map; import java.util.Random; -import java.util.Set; import java.util.concurrent.TimeUnit; -import static org.junit.Assert.assertEquals; - 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"; @@ -46,6 +41,7 @@ public class BGChronicleTest { /** * 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 @@ -100,36 +96,6 @@ public class BGChronicleTest { } - @Test(timeout = 10000) - public void testBufferOverFlowPutAllAndEntrySet() throws IOException, InterruptedException { - int SIZE = 10; - int port = 9050; - try (ChronicleMap serverMap = ChronicleMapBuilder - .of(Integer.class, CharSequence.class) - .entries(SIZE) - .replication((byte) 2, TcpTransportAndNetworkConfig.of(port)) - .create()) { - try (ChronicleMap statelessMap = ChronicleMapBuilder - .of(Integer.class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", port)) - .create()) { - Map payload = new HashMap(); - - for (int i = 0; i < SIZE; i++) { - payload.put(i, "some value=" + i); - } - - statelessMap.putAll(payload); - - Set> entries = statelessMap.entrySet(); - - Map.Entry next = entries.iterator().next(); - assertEquals("some value=" + next.getKey(), next.getValue()); - - assertEquals(SIZE, entries.size()); - } - } - } /** @@ -164,6 +130,9 @@ public class BGChronicleTest { o.startChronicleMapClient(host, port, MAX_NUMBER_OF_EATERIES); } else { System.out.println("client if OFF"); + for (;;){ + Thread.sleep(1000); + } } } finally { From ca98f659fdc4ac8b094ea9e086f10ebb1956441d Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 15:19:58 +0000 Subject: [PATCH 33/35] added the following to prevent writting to a socket that is not bound - if (!clientChannel.finishConnect() || !clientChannel.socket().isBound()) --- .../chronicle/map/StatelessChronicleMap.java | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index e5f886d8b..2c88efaff 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -287,9 +287,12 @@ class StatelessChronicleMap implements ChronicleMap, 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); } @@ -983,13 +986,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long inTransactionId = bytes.readLong(); if (inTransactionId != transactionId) { - LOG.error("", new IllegalStateException("Skipped Message with transaction-id=" + + 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")); + "stateless client and terminated abruptly before the message has been " + + "returned from the server")); - blockingFetch(timeoutTime,transactionId); + blockingFetch(timeoutTime, transactionId); } if (isException) { From 2c4cfa32ec48a581f30b08b4fb1d7c15f9f5ea06 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 16:15:48 +0000 Subject: [PATCH 34/35] added the replicating client --- src/test/java/eg/BGChronicleTest.java | 141 +++++++++++++++++++++----- 1 file changed, 117 insertions(+), 24 deletions(-) diff --git a/src/test/java/eg/BGChronicleTest.java b/src/test/java/eg/BGChronicleTest.java index d7bca7df5..229382dfe 100644 --- a/src/test/java/eg/BGChronicleTest.java +++ b/src/test/java/eg/BGChronicleTest.java @@ -24,7 +24,7 @@ public class BGChronicleTest { public ChronicleMap startChronicleMapServer(int port) throws InterruptedException, IOException { - TcpTransportAndNetworkConfig map2Config = TcpTransportAndNetworkConfig.of(port) + TcpTransportAndNetworkConfig serverConfig = TcpTransportAndNetworkConfig.of(port) .heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS).name("serverMap") .autoReconnectedUponDroppedConnection(true); @@ -35,13 +35,13 @@ public class BGChronicleTest { .valueSize(11) .putReturnsNull(true) .removeReturnsNull(true) - .replication((byte) 1, map2Config) + .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 + * 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 @@ -88,20 +88,76 @@ public class BGChronicleTest { } } - private void logMessagesPerSecond(double numberOfMessages, long start) { - double messagesPerSecond = numberOfMessages / (double) TimeUnit.NANOSECONDS - .toSeconds(System.nanoTime() - start); - System.out.println("messages per seconds " + messagesPerSecond); + /** + * 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 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 servereg.BGChronicleTest server + * for server - eg.BGChronicleTest server for client - eg.BGChronicleTest client * * @param args * @throws IOException @@ -113,26 +169,50 @@ public class BGChronicleTest { ChronicleMap serverMap = null; try { - if (args.length == 0 || (args.length > 0 && args[0].equalsIgnoreCase("server"))) { - int port = args.length > 1 && args[1] != null ? Integer.valueOf(args[1]) : - DEFAULT_PORT; + // loop-back + if (args.length == 0) { + serverMap = o.startChronicleMapServer(DEFAULT_PORT); + o.startChronicleMapClient(DEFAULT_HOST, DEFAULT_PORT, MAX_NUMBER_OF_EATERIES); - serverMap = o.startChronicleMapServer(port); - System.out.println("server ON"); - } else - System.out.println("server if OFF"); - - if (args.length == 0 || (args.length > 1 && args[0].equalsIgnoreCase("client"))) { - String host = args.length > 1 && args[1] != null ? args[1] : DEFAULT_HOST; - int port = args.length > 2 && args[2] != null ? Integer.valueOf(args[2]) : - DEFAULT_PORT; - System.out.println("client ON...."); - o.startChronicleMapClient(host, 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 (;;){ + for (; ; ) { Thread.sleep(1000); } + } } finally { @@ -142,6 +222,19 @@ public class BGChronicleTest { } + 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(); } From 8d28d0194daaeb09c6aa954a02c4b6063bc2032e Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 3 Dec 2014 17:40:15 +0000 Subject: [PATCH 35/35] HCOLL-230 - I flipped around the exception flag and the transaction id --- .../map/AbstractChannelReplicator.java | 1 + .../chronicle/map/StatelessChronicleMap.java | 3 ++- .../openhft/chronicle/map/TcpReplicator.java | 21 +++++++++++-------- 3 files changed, 15 insertions(+), 10 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 37b6b2e93..80267adbf 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -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); diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 2c88efaff..789169a58 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -982,8 +982,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // block until we have received all the bytes in this chunk receive(size, timeoutTime); - final boolean isException = bytes.readBoolean(); + final long inTransactionId = bytes.readLong(); + final boolean isException = bytes.readBoolean(); if (inTransactionId != transactionId) { LOG.error("", new IllegalStateException("Skipped Message with transaction-id=" + diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index cd24a9757..9d78cb591 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1520,7 +1520,7 @@ class StatelessServerConnector { writeException(writer, e); - writeSizeAndFlags(sizeLocation, true, writer); + writeSizeAndFlags(sizeLocation+SIZE_OF_TRANSACTION_ID, true, writer); return null; } @@ -1845,17 +1845,19 @@ class StatelessServerConnector { } 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) { @@ -1887,10 +1889,11 @@ class StatelessServerConnector { 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); @@ -1909,12 +1912,12 @@ class StatelessServerConnector { // 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