From bf65306246271fa0b12f3adad4c9a3889e2272ef Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 20 Nov 2014 12:18:07 +0000 Subject: [PATCH] HCOLL-134 returned the remoteIdentifier back to a BYTE and improved the MapByName --- .../chronicle/hash/ChronicleHashBuilder.java | 368 ++++++++--------- .../map/AbstractChronicleMapBuilder.java | 2 +- .../chronicle/map/ChannelProvider.java | 48 ++- .../chronicle/map/ChronicleMapBuilder.java | 4 +- .../chronicle/map/ChronicleMapBuilderI.java | 2 +- .../net/openhft/chronicle/map/Replica.java | 2 +- .../chronicle/map/ReplicatedChronicleMap.java | 2 +- .../map/ReplicationHubFindByName.java | 22 +- .../openhft/chronicle/map/TcpReplicator.java | 4 + .../chronicle/set/ChronicleSetBuilder.java | 13 +- .../openhft/chronicle/map/MapByNameTest.java | 17 +- .../chronicle/map/example/ChannelExample.java | 165 -------- .../map/example/ChannelReplicationTest.java | 375 ++++++++++++++++++ 13 files changed, 627 insertions(+), 397 deletions(-) delete mode 100644 src/test/java/net/openhft/chronicle/map/example/ChannelExample.java create mode 100644 src/test/java/net/openhft/chronicle/map/example/ChannelReplicationTest.java diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index e07a2e19b..c84f2cf82 100755 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -22,7 +22,7 @@ import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.hash.replication.TimeProvider; import net.openhft.chronicle.hash.serialization.*; -import net.openhft.chronicle.map.*; +import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.set.ChronicleSet; import net.openhft.chronicle.set.ChronicleSetBuilder; import net.openhft.lang.io.Bytes; @@ -45,36 +45,32 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; /** - * This interface defines the meaning of configurations, common to - * {@link net.openhft.chronicle.map.OnHeapUpdatableChronicleMapBuilder} and - * {@link ChronicleSetBuilder}, i. e. Chronicle hash container configurations. + * This interface defines the meaning of configurations, common to {@link net.openhft.chronicle.map.OnHeapUpdatableChronicleMapBuilder} + * and {@link ChronicleSetBuilder}, i. e. Chronicle hash container configurations. * - *

{@code ChronicleHashBuilder} is mutable. Configuration methods mutate the builder and return - * the builder itself back to support chaining pattern, rather than the builder copies with - * the corresponding configuration changed. To make an independent configuration, {@linkplain - * #clone} the builder. + *

{@code ChronicleHashBuilder} is mutable. Configuration methods mutate the builder and return the + * builder itself back to support chaining pattern, rather than the builder copies with the corresponding + * configuration changed. To make an independent configuration, {@linkplain #clone} the builder. * * @param the type of keys in hash containers, created by this builder - * @param the container type, created by this builder, i. e. {@link ChronicleMap} or {@link - * ChronicleSet} - * @param the concrete builder type, i. e. {@link net.openhft.chronicle.map.ChronicleMapOnHeapUpdatableBuilder} or {@link - * ChronicleSetBuilder} + * @param the container type, created by this builder, i. e. {@link ChronicleMap} or {@link ChronicleSet} + * @param the concrete builder type, i. e. {@link net.openhft.chronicle.map.ChronicleMapOnHeapUpdatableBuilder} + * or {@link ChronicleSetBuilder} */ public interface ChronicleHashBuilder> extends Cloneable { /** - * Clones this builder. Useful for configuration persisting, because {@code - * ChronicleHashBuilder}s are mutable and changed on each configuration method call. Original - * and cloned builders are independent. + * Clones this builder. Useful for configuration persisting, because {@code ChronicleHashBuilder}s are + * mutable and changed on each configuration method call. Original and cloned builders are independent. * * @return a new clone of this builder */ B clone(); /** - * Set minimum number of segments in hash containers, constructed by this builder. See - * concurrencyLevel in {@link ConcurrentHashMap}. + * Set minimum number of segments in hash containers, constructed by this builder. See concurrencyLevel in + * {@link ConcurrentHashMap}. * * @param minSegments the minimum number of segments in containers, constructed by this builder * @return this builder object back @@ -82,14 +78,13 @@ public interface ChronicleHashBuilderIf key size varies moderately, specify the size higher than average, but lower than the - * maximum possible, to minimize average memory overuse. If key size varies in a wide range, - * it's better to use {@linkplain #entrySize(int) entry size} in "chunk" mode and configure it - * directly. + *

If key size varies moderately, specify the size higher than average, but lower than the maximum + * possible, to minimize average memory overuse. If key size varies in a wide range, it's better to use + * {@linkplain #entrySize(int) entry size} in "chunk" mode and configure it directly. * * @param keySize number of bytes, taken by serialized form of keys * @return this builder back @@ -99,15 +94,15 @@ public interface ChronicleHashBuilderIf keys are of boxed primitive type or {@link Byteable} subclass, i. e. if key size is - * known statically, it is automatically accounted and this method shouldn't be called. + *

If keys are of boxed primitive type or {@link Byteable} subclass, i. e. if key size is known + * statically, it is automatically accounted and this method shouldn't be called. * - *

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

If key size varies, method {@link #keySize(int)} or {@link #entrySize(int)} should be called instead + * of this one. * * @param sampleKey the sample key * @return this builder back @@ -116,45 +111,42 @@ public interface ChronicleHashBuilder{@link ChronicleMap} and {@link ChronicleSet} store their data off-heap, so it is required - * to serialize key (and values, in {@code ChronicleMap} case) (unless they are direct {@link - * Byteable} instances). Serialized key bytes (+ serialized value bytes, in {@code ChronicleMap} - * case) + some metadata bytes comprise "entry space", which {@code ChronicleMap} or {@code - * ChronicleSet} should allocate. So entry size is a minimum allocation portion in the - * hash containers, created by this builder. E. g. if entry size is 100, the created container - * could only allocate 100, 200, 300... bytes for an entry. If say 150 bytes of entry space are - * required by the entry, 200 bytes will be allocated, 150 used and 50 wasted. To minimize - * memory overuse and improve speed, you should pay decent attention to this configuration. + *

{@link ChronicleMap} and {@link ChronicleSet} store their data off-heap, so it is required to + * serialize key (and values, in {@code ChronicleMap} case) (unless they are direct {@link Byteable} + * instances). Serialized key bytes (+ serialized value bytes, in {@code ChronicleMap} case) + some + * metadata bytes comprise "entry space", which {@code ChronicleMap} or {@code ChronicleSet} should + * allocate. So entry size is a minimum allocation portion in the hash containers, created by this + * builder. E. g. if entry size is 100, the created container could only allocate 100, 200, 300... bytes + * for an entry. If say 150 bytes of entry space are required by the entry, 200 bytes will be allocated, + * 150 used and 50 wasted. To minimize memory overuse and improve speed, you should pay decent attention + * to this configuration. * - *

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

    There 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)} - * 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 - * maximum possible. The idea is to have most (90% or more) entries to fit a single "entry size" - * with moderate memory waste (10-20% on average), rest 10% or less of entries should take 2 - * "entry sizes", thus with ~50% memory overuse.
    3. Key (and/or value size, in {@code - * ChronicleMap} case) varies in a wide range. Then it's best to use entry size configuration in - * chunk mode. Specify entry size so that most entries should take from 5 to several - * dozens of "chunks". With this approach, average memory waste should be very low. + * methods, and you will experience no memory waste at all.
    4. 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 maximum possible. The idea is + * to have most (90% or more) entries to fit a single "entry size" with moderate memory waste (10-20% on + * average), rest 10% or less of entries should take 2 "entry sizes", thus with ~50% memory overuse.
      5. + *
    5. Key (and/or value size, in {@code ChronicleMap} case) varies in a wide range. Then it's best to use + * entry size configuration in chunk mode. Specify entry size so that most entries should take from + * 5 to several dozens of "chunks". With this approach, average memory waste should be very low. * - *

      However, remember that

      • Operations with entries that span several "entry sizes" - * are a bit slower, than entries which take a single "entry size". That is why "chunk" approach - * is not recommended, when key (and/or value size) varies moderately.
      • The maximum - * number of chunks could be taken by an entry is 64. {@link IllegalArgumentException} is thrown - * on attempt to insert too large entry, compared to the configured or computed entry size.
        • - *
      • The number of "entries" {@linkplain #entries(long) in the whole map} and {@linkplain - * #actualEntriesPerSegment(long) per segment} is actually the number "chunks".
      + *

      However, remember that

      • Operations with entries that span several "entry sizes" are a bit + * slower, than entries which take a single "entry size". That is why "chunk" approach is not recommended, + * when key (and/or value size) varies moderately.
      • The maximum number of chunks could be taken by + * an entry is 64. {@link IllegalArgumentException} is thrown on attempt to insert too large entry, + * compared to the configured or computed entry size.
      • The number of "entries" {@linkplain + * #entries(long) in the whole map} and {@linkplain #actualEntriesPerSegment(long) per segment} is + * actually the number "chunks".
      * - *

      Example: if values in your {@code ChronicleMap} are adjacency lists of some social graph, - * where nodes are represented as {@code long} ids, and adjacency lists are serialized in - * efficient manner, for example as {@code long[]} arrays. Typical number of connections is - * 100-300, maximum is 3000. In this case entry size of + *

      Example: if values in your {@code ChronicleMap} are adjacency lists of some social graph, where + * nodes are represented as {@code long} ids, and adjacency lists are serialized in efficient manner, for + * example as {@code long[]} arrays. Typical number of connections is 100-300, maximum is 3000. In this + * case entry size of * 50 * (8 bytes for each id) = 400 bytes would be a good choice: 

      {@code
      * Map socialGraph = ChronicleMapOnHeapUpdatableBuilder
      *     .of(Long.class, long[].class)
@@ -163,9 +155,8 @@ public interface ChronicleHashBuilder
-     * It is minimum possible (because 3000 friends / 50 friends = 60 is close to 64 "max chunks by
-     * single entry" limit, and ensures moderate average memory overuse (not more than 20%).
      6. - *
    + * It is minimum possible (because 3000 friends / 50 friends = 60 is close to 64 "max chunks by single + * entry" limit, and ensures moderate average memory overuse (not more than 20%).
* * @param entrySize the "chunk size" in bytes * @return this builder back @@ -174,32 +165,29 @@ public interface ChronicleHashBuilder
  • If key size (and value size, in {@code ChronicleMap} case) is constant, this number - * is equal to the maximum number of entries (because each entry takes exactly one "entry size" - * memory unit).
  • 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) value}/{@linkplain #entrySize(int) entry} sizes.
    • - *
  • 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 #entrySize(int)} method.
    • + *
    1. If key size (and value size, in {@code ChronicleMap} case) is constant, this number is equal + * to the maximum number of entries (because each entry takes exactly one "entry size" memory unit).
      2. + *
    2. 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) + * value}/{@linkplain #entrySize(int) entry} sizes.
    3. 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 #entrySize(int)} method.
      4. + *
    * - *

    You shouldn't put additional margin over the number, computed according the rules above. - * This bad practice was popularized by {@link HashMap#HashMap(int)} and {@link - * HashSet#HashSet(int)} constructors, which accept "capacity", that should be multiplied by - * "load factor" to obtain actual maximum expected number of entries. {@code ChronicleMap} and - * {@code ChronicleSet} don't have a notion of load factor. + *

    You shouldn't put additional margin over the number, computed according the rules above. This bad + * practice was popularized by {@link HashMap#HashMap(int)} and {@link HashSet#HashSet(int)} constructors, + * which accept "capacity", that should be multiplied by "load factor" to obtain actual maximum expected + * number of entries. {@code ChronicleMap} and {@code ChronicleSet} don't have a notion of load factor. * *

    Default value is 2^20 (~ 1 million). * - * @param entries maximum size of the created maps, in memory allocation units, so-called "entry - * size" + * @param entries maximum size of the created maps, in memory allocation units, so-called "entry size" * @return this builder back * @see #entrySize(int) */ @@ -210,18 +198,17 @@ public interface ChronicleHashBuilderDefault lock time out is 2 seconds. * - * @param lockTimeOut new lock timeout for segments of containers created by this builder, in - * the given time units + * @param lockTimeOut new lock timeout for segments of containers created by this builder, in the given + * time units * @param unit time unit of the given lock timeout * @return this builder back */ @@ -232,8 +219,8 @@ public interface ChronicleHashBuilderDefault time provider is {@link TimeProvider#SYSTEM}. * @@ -244,17 +231,15 @@ public interface ChronicleHashBuilderDefault {@code BytesMarshallerFactory} is an instance of {@link - * VanillaBytesMarshallerFactory}. This is a convenience configuration method, it has no effect - * on the resulting hash containers, if {@linkplain #keyMarshaller(BytesMarshaller) custom data - * marshallers} are configured, data types extends one of specific serialization interfaces, - * recognized by this builder (e. g. {@code Externalizable} or {@code BytesMarshallable}), or - * {@code ObjectSerializer} is configured. + *

    Default {@code BytesMarshallerFactory} is an instance of {@link VanillaBytesMarshallerFactory}. This + * is a convenience configuration method, it has no effect on the resulting hash containers, if + * {@linkplain #keyMarshaller(BytesMarshaller) custom data marshallers} are configured, data types extends + * one of specific serialization interfaces, recognized by this builder (e. g. {@code Externalizable} or + * {@code BytesMarshallable}), or {@code ObjectSerializer} is configured. * * @param bytesMarshallerFactory the marshaller factory to be used with the default {@code * ObjectSerializer}, i. e. {@code BytesMarshallableSerializer} @@ -264,18 +249,18 @@ public interface ChronicleHashBuilderDefault serializer is {@link BytesMarshallableSerializer}, configured with the specified - * or default {@link #bytesMarshallerFactory(BytesMarshallerFactory) BytesMarshallerFactory}. + *

    Default serializer is {@link BytesMarshallableSerializer}, configured with the specified or default + * {@link #bytesMarshallerFactory(BytesMarshallerFactory) BytesMarshallerFactory}. * - * @param objectSerializer the serializer used to serialize loosely typed or nullable data if - * custom marshaller is not configured + * @param objectSerializer the serializer used to serialize loosely typed or nullable data if custom + * marshaller is not configured * @return this builder back * @see #bytesMarshallerFactory(BytesMarshallerFactory) * @see #keyMarshaller(BytesMarshaller) @@ -283,8 +268,8 @@ public interface ChronicleHashBuilderthe + * Configures the {@code BytesMarshaller} used to serialize/deserialize keys to/from off-heap memory in + * hash containers, created by this builder. See the * section about serialization in ChronicleMap manual for more information. * * @param keyMarshaller the marshaller used to serialize keys @@ -295,17 +280,15 @@ public interface ChronicleHashBuilder keyMarshaller); /** - * Configures the marshallers, used to serialize/deserialize keys to/from off-heap - * memory in hash containers, created by this builder. See - * the section about - * serialization in ChronicleMap manual for more information. + * Configures the marshallers, used to serialize/deserialize keys to/from off-heap memory in hash + * containers, 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 #keyMarshaller(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 #keyMarshaller(BytesMarshaller)}. * - *

    Passing {@link BytesInterop} (which is a subinterface of {@link BytesWriter}) as the - * first argument is supported, and even more advantageous from performance perspective. + *

    Passing {@link BytesInterop} (which is a subinterface of {@link BytesWriter}) as the first argument + * is supported, and even more advantageous from performance perspective. * * @param keyWriter the new key object → {@link Bytes} writer (interop) strategy * @param keyReader the new {@link Bytes} → key object reader strategy @@ -315,70 +298,67 @@ public interface ChronicleHashBuilder keyWriter, @NotNull BytesReader keyReader); /** - * Configures the marshaller used to serialize actual key sizes to off-heap memory - * in hash containers, created by this builder. + * Configures the marshaller used to serialize actual key sizes to off-heap memory in hash containers, + * created by this builder. * - *

    Default key size marshaller is so-called {@linkplain SizeMarshallers#stopBit() - * stop bit encoding marshalling}. If {@linkplain #constantKeySizeBySample(Object) constant key - * size} is configured, or defaulted if the key type is always constant and {@code - * ChronicleHashBuilder} implementation knows about it, this configuration takes no effect, - * because a special {@link SizeMarshaller} implementation, which doesn't actually do any - * marshalling, and just returns the known constant size on {@link SizeMarshaller#readSize( - * Bytes)} calls, is used instead of any {@code SizeMarshaller} configured using this method. + *

    Default key size marshaller is so-called {@linkplain SizeMarshallers#stopBit() stop bit encoding + * marshalling}. If {@linkplain #constantKeySizeBySample(Object) constant key size} is configured, or + * defaulted if the key type is always constant and {@code ChronicleHashBuilder} implementation knows + * about it, this configuration takes no effect, because a special {@link SizeMarshaller} implementation, + * which doesn't actually do any marshalling, and just returns the known constant size on {@link + * SizeMarshaller#readSize(Bytes)} calls, is used instead of any {@code SizeMarshaller} configured using + * this method. * - * @param keySizeMarshaller the new marshaller, used to serialize actual key sizes to off-heap - * memory + * @param keySizeMarshaller the new marshaller, used to serialize actual key sizes to off-heap memory * @return this builder back */ B keySizeMarshaller(@NotNull SizeMarshaller keySizeMarshaller); /** - * Configures factory which is used to create a new key instance, if key class is either - * {@link Byteable}, {@link BytesMarshallable} or {@link Externalizable} subclass in maps, - * created by this builder. If {@linkplain #keyMarshaller(BytesMarshaller) custom key - * marshaller} is configured, this configuration is unused, because it is incapsulated in - * {@link BytesMarshaller#read(Bytes)} method (without provided instance to read the data into), - * i. e. it's is the user-side responsibility. + * Configures factory which is used to create a new key instance, if key class is either {@link Byteable}, + * {@link BytesMarshallable} or {@link Externalizable} subclass in maps, created by this builder. If + * {@linkplain #keyMarshaller(BytesMarshaller) custom key marshaller} is configured, this configuration is + * unused, because it is incapsulated in {@link BytesMarshaller#read(Bytes)} method (without provided + * instance to read the data into), i. e. it's is the user-side responsibility. * - *

    Default key deserialization factory is {@link NewInstanceObjectFactory}, which creates - * a new key instance using {@link Class#newInstance()} default constructor. You could provide - * an {@link AllocateInstanceObjectFactory}, which uses {@code Unsafe.allocateInstance(Class)} - * (you might want to do this for better performance or if you don't want to initialize fields), - * or a factory which calls a key class constructor with some arguments, or a factory which - * internally delegates to instance pool or {@link ThreadLocal}, to reduce allocations. + *

    Default key deserialization factory is {@link NewInstanceObjectFactory}, which creates a new key + * instance using {@link Class#newInstance()} default constructor. You could provide an {@link + * AllocateInstanceObjectFactory}, which uses {@code Unsafe.allocateInstance(Class)} (you might want to do + * this for better performance or if you don't want to initialize fields), or a factory which calls a key + * class constructor with some arguments, or a factory which internally delegates to instance pool or + * {@link ThreadLocal}, to reduce allocations. * - * @param keyDeserializationFactory the key factory used to produce instances to deserialize - * data in + * @param keyDeserializationFactory the key factory used to produce instances to deserialize data in * @return this builder back - * @throws IllegalStateException if custom key marshaller is specified or key class is not - * either {@code Byteable}, {@code BytesMarshallable} or {@code Externalizable} + * @throws IllegalStateException if custom key marshaller is specified or key class is not either {@code + * Byteable}, {@code BytesMarshallable} or {@code Externalizable} */ B keyDeserializationFactory(@NotNull ObjectFactory keyDeserializationFactory); /** - * Specifies that key objects, queried with the hash containers, created by this builder, are - * inherently immutable. Keys in {@link ChronicleMap} or {@link ChronicleSet} are not required - * to be immutable, as in ordinary {@link Map} or {@link Set} implementations, because they are - * serialized off-heap. However, {@code ChronicleMap} and {@code ChronicleSet} implementations - * can benefit from the knowledge that keys are not mutated between queries. + * Specifies that key objects, queried with the hash containers, created by this builder, are inherently + * immutable. Keys in {@link ChronicleMap} or {@link ChronicleSet} are not required to be immutable, as in + * ordinary {@link Map} or {@link Set} implementations, because they are serialized off-heap. However, + * {@code ChronicleMap} and {@code ChronicleSet} implementations can benefit from the knowledge that keys + * are not mutated between queries. * - *

    By default, {@code ChronicleHashBuilder}s detects immutability automatically only for very - * few standard JDK types (for example, for {@link String}), it is not recommended to rely on - * {@code ChronicleHashBuilder} to be smart enough about this. + *

    By default, {@code ChronicleHashBuilder}s detects immutability automatically only for very few + * standard JDK types (for example, for {@link String}), it is not recommended to rely on {@code + * ChronicleHashBuilder} to be smart enough about this. * * @return this builder back */ B immutableKeys(); /** - * Configures replication of the hash containers, created by this builder. - * See - * the section about replication in ChronicleMap manual for more information. + * Configures replication of the hash containers, created by this builder. See the section about replication in + * ChronicleMap manual for more information. * *

    By default, hash containers, created by this builder doesn't replicate their data. * - *

    This method call overrides all previous replication configurations of this builder, made - * either by this method or {@link #replication(byte, TcpTransportAndNetworkConfig)} shortcut method. + *

    This method call overrides all previous replication configurations of this builder, made either by + * this method or {@link #replication(byte, TcpTransportAndNetworkConfig)} shortcut method. * * @param replication the replication config * @return this builder back @@ -388,10 +368,9 @@ public interface ChronicleHashBuilder instance(); /** - * Creates a new hash container, storing it's data in off-heap memory, not mapped to any file. - * On {@link ChronicleHash#close()} called on the returned container, or after the container - * object is collected during GC, or on JVM shutdown the off-heap memory used by the returned - * container is freed. + * Creates a new hash container, storing it's data in off-heap memory, not mapped to any file. On {@link + * ChronicleHash#close()} called on the returned container, or after the container object is collected + * during GC, or on JVM shutdown the off-heap memory used by the returned container is freed. * *

    This method is a shortcut for {@code instance().create()}. * * @return a new off-heap hash container - * @throws IOException if any IO error relates to off-heap memory allocation, - * or establishing replication connections, occurs + * @throws IOException if any IO error relates to off-heap memory allocation, or establishing replication + * connections, occurs * @see #createPersistedTo(File) * @see #instance() */ C create() throws IOException; /** - * Opens a hash container residing the specified file, or creates a new one if the file not yet - * exists and maps its off-heap memory to the file. All changes to the map are persisted to disk - * (this is an operating system guarantee) independently from JVM process lifecycle. + * Opens a hash container residing the specified file, or creates a new one if the file not yet exists and + * maps its off-heap memory to the file. All changes to the map are persisted to disk (this is an + * operating system guarantee) independently from JVM process lifecycle. * - *

    Multiple containers could give access to the same data simultaneously, either inside a - * single JVM or across processes. Access is synchronized correctly across all instances, i. e. - * hash container mapping the data from the first JVM isn't able to modify the data, - * concurrently accessed from the second JVM by another hash container instance, mapping the - * same data. + *

    Multiple containers could give access to the same data simultaneously, either inside a single JVM or + * across processes. Access is synchronized correctly across all instances, i. e. hash container mapping + * the data from the first JVM isn't able to modify the data, concurrently accessed from the second JVM by + * another hash container instance, mapping the same data. * - *

    On container's {@link ChronicleHash#close() close()} the data isn't removed, it remains on - * disk and available to be opened again (given the same file name) or during different JVM - * run. + *

    On container's {@link ChronicleHash#close() close()} the data isn't removed, it remains on disk and + * available to be opened again (given the same file name) or during different JVM run. * *

    This method is shortcut for {@code instance().persistedTo(file).create()}. * - * @param file the file with existing hash container or a desired location of a new off-heap - * persisted hash container + * @param file the file with existing hash container or a desired location of a new off-heap persisted + * hash container * @return a hash container mapped to the given file - * @throws IOException if any IO error, related to off-heap memory allocation or file mapping, - * or establishing replication connections, occurs + * @throws IOException if any IO error, related to off-heap memory allocation or file mapping, or + * establishing replication connections, occurs * @see ChronicleHash#file() * @see ChronicleHash#close() * @see #create() @@ -450,4 +426,6 @@ public interface ChronicleHashBuilder> implements Cloneable, ChronicleHashBuilder, B>, ChronicleMapBuilderI { - static final short UDP_REPLICATION_MODIFICATION_ITERATOR_ID = 128; + static final byte UDP_REPLICATION_MODIFICATION_ITERATOR_ID = (byte)128; private static final Bytes EMPTY_BYTES = new ByteBufferBytes(ByteBuffer.allocate(0)); private static final int DEFAULT_KEY_OR_VALUE_SIZE = 120; private static final int MAX_SEGMENTS = (1 << 30); diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 9a3571b0e..d07e76a4d 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -112,9 +112,12 @@ final class ChannelProvider implements Closeable { channelDataLock.readLock().lock(); try { final int chronicleId = (int) source.readStopBit(); - if (chronicleId < chronicleChannels.length) - channelEntryExternalizables[chronicleId].readExternalEntry(source); - else + if (chronicleId < chronicleChannels.length) { + + // this channel is has not currently been created so it updates will be ignored + if (channelEntryExternalizables[chronicleId] != null) + channelEntryExternalizables[chronicleId].readExternalEntry(source); + } else LOG.info("skipped entry with chronicleId=" + chronicleId + ", "); } finally { channelDataLock.readLock().unlock(); @@ -131,7 +134,8 @@ final class ChannelProvider implements Closeable { @Override public ModificationIterator acquireModificationIterator( - final short remoteIdentifier, final ModificationNotifier notifier) { + final byte remoteIdentifier, final ModificationNotifier notifier) { + channelDataLock.writeLock().lock(); try { final ModificationIterator result = modificationIterator.get(remoteIdentifier); @@ -248,6 +252,7 @@ final class ChannelProvider implements Closeable { private final Set replicators = new CopyOnWriteArraySet(); private volatile boolean isClosed = false; + private final SystemQueue systemMessageQueue; private ChannelProvider(ReplicationHub hub) { localIdentifier = hub.identifier(); @@ -268,7 +273,7 @@ final class ChannelProvider implements Closeable { } } }; - SystemQueue systemMessageQueue = new SystemQueue( + systemMessageQueue = new SystemQueue( systemModificationIteratorBitSet, systemModificationIterator, systemMessageHandler); add((short) 0, systemMessageQueue.asReplica, systemMessageQueue.asEntryExternalizable); } @@ -296,10 +301,15 @@ final class ChannelProvider implements Closeable { * called whenever we receive a bootstrap message */ private void onBootstrapMessage(Bytes bytes) { - final short remoteIdentifier = bytes.readByte(); + final byte remoteIdentifier = bytes.readByte(); final int chronicleChannel = bytes.readUnsignedShort(); final long lastModificationTime = bytes.readLong(); + if (LOG.isDebugEnabled()) + LOG.debug("received bootstrap message received for localIdentifier=" + this.localIdentifier + ", " + + "remoteIdentifier=" + remoteIdentifier + ",chronicleChannel=" + chronicleChannel + "," + + "lastModificationTime=" + lastModificationTime); + // this could be null if one node has a chronicle channel before the other if (chronicleChannels[chronicleChannel] != null) { chronicleChannels[chronicleChannel].acquireModificationIterator(remoteIdentifier, NOP) @@ -307,7 +317,9 @@ final class ChannelProvider implements Closeable { } } - private ByteBufferBytes toBootstrapMessage(int chronicleChannel, final long lastModificationTime) { + private static ByteBufferBytes toBootstrapMessage(int chronicleChannel, final long lastModificationTime, final byte localIdentifier) { + + final ByteBufferBytes writeBuffer = new ByteBufferBytes(ByteBuffer.allocate(1 + 1 + 2 + 8)); writeBuffer.writeByte(BOOTSTRAP_MESSAGE); writeBuffer.writeByte(localIdentifier); @@ -320,8 +332,9 @@ final class ChannelProvider implements Closeable { private void add(int chronicleChannel, Replica replica, @NotNull EntryExternalizable entryExternalizable) { - - LOG.info("adding chronicleChannel=" + chronicleChannel + ",entryExternalizable=" + entryExternalizable); + if (LOG.isDebugEnabled()) + LOG.debug("adding chronicleChannel=" + chronicleChannel + ",entryExternalizable=" + + entryExternalizable); channelDataLock.writeLock().lock(); try { if (chronicleChannels[chronicleChannel] != null) { @@ -336,13 +349,21 @@ final class ChannelProvider implements Closeable { if (chronicleChannel == 0) return; + // send bootstap message + for (int i = (int) systemModificationIteratorBitSet.nextSetBit(0); i > 0; i = (int) systemModificationIteratorBitSet.nextSetBit(i + 1)) { byte remoteIdentifier = (byte) i; final long lastModificationTime = replica.lastModificationTime(remoteIdentifier); final ByteBufferBytes message = - toBootstrapMessage(chronicleChannel, lastModificationTime); + toBootstrapMessage(chronicleChannel, lastModificationTime, localIdentifier); systemModificationIterator.get(remoteIdentifier).addPayload(message); + + if (LOG.isDebugEnabled()) + LOG.debug("sending bootstrap message received for localIdentifier=" + localIdentifier + ", " + + "remoteIdentifier=" + remoteIdentifier + ",chronicleChannel=" + chronicleChannel + "," + + "lastModificationTime=" + lastModificationTime); + } } finally { channelDataLock.writeLock().unlock(); @@ -384,19 +405,20 @@ final class ChannelProvider implements Closeable { * used to send system messages such as bootstrap from one remote node to another, it also can be used in * a broadcast context */ - static class SystemQueue { + class SystemQueue { final Replica asReplica = new Replica() { @Override public byte identifier() { - return 0; + throw new UnsupportedOperationException(); } @Override public ModificationIterator acquireModificationIterator( - final short remoteIdentifier, final ModificationNotifier modificationNotifier) { + final byte remoteIdentifier, final ModificationNotifier modificationNotifier) { + final ModificationIterator result = systemModificationIterator.get(remoteIdentifier); diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 0ad9ed198..d956e13db 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -1,5 +1,6 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.hash.ChronicleHashBuilder; import net.openhft.chronicle.hash.ChronicleHashErrorListener; import net.openhft.chronicle.hash.ChronicleHashInstanceConfig; import net.openhft.chronicle.hash.StatelessClientConfig; @@ -30,7 +31,8 @@ import java.util.concurrent.TimeUnit; * @see AbstractChronicleMapBuilder * @see net.openhft.chronicle.hash.ChronicleHashBuilder */ -public final class ChronicleMapBuilder implements ChronicleMapBuilderI { +public final class ChronicleMapBuilder implements ChronicleMapBuilderI, ChronicleHashBuilder, ChronicleMapBuilder> { final ChronicleMapBuilderI delegate; diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java index cdf5df893..bfb940eb9 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java @@ -24,7 +24,7 @@ import java.util.concurrent.TimeUnit; /** * @author Rob Austin. */ -interface ChronicleMapBuilderI extends Serializable { +public interface ChronicleMapBuilderI extends Serializable { /** diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 1041f7322..0ca766f4b 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -53,7 +53,7 @@ interface Replica extends Closeable { * @return the ModificationIterator dedicated for replication to the remote node with the given identifier * @see #identifier() */ - ModificationIterator acquireModificationIterator(short remoteIdentifier, + ModificationIterator acquireModificationIterator(byte remoteIdentifier, ModificationNotifier modificationNotifier); /** diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 49ee7fbe2..9c0dd6062 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -362,7 +362,7 @@ final class ReplicatedChronicleMap, @Override public Replica.ModificationIterator acquireModificationIterator - (short remoteIdentifier, + (byte remoteIdentifier, @NotNull final ModificationNotifier modificationNotifier) { return modificationDelegator diff --git a/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java b/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java index 523f477fd..8f404c1a9 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHash; +import net.openhft.chronicle.hash.ChronicleHashBuilder; import net.openhft.chronicle.hash.ChronicleHashInstanceConfig; import net.openhft.chronicle.hash.FindByName; import net.openhft.chronicle.hash.replication.ReplicationChannel; @@ -37,7 +38,7 @@ import static net.openhft.chronicle.map.ChronicleMapBuilder.of; /** * @author Rob Austin. */ -class ReplicationHubFindByName implements FindByName { +class ReplicationHubFindByName implements FindByName { public static final Logger LOG = LoggerFactory.getLogger(ReplicationHubFindByName.class.getName()); public static final int MAP_BY_NAME_CHANNEL = 1; @@ -47,8 +48,8 @@ class ReplicationHubFindByName implements FindByName { private final ReplicationHub replicationHub; - public static class ChronicleMapBuilderWithChannelId implements Serializable { - ChronicleMapBuilder chronicleMapBuilder; + public static class ChronicleMapBuilderWithChannelId implements Serializable { + ChronicleHashBuilder, ?> chronicleMapBuilder; int channelId; } @@ -103,21 +104,24 @@ class ReplicationHubFindByName implements FindByName { .instance() .replicatedViaChannel(channel) .create(); + if (LOG.isDebugEnabled()) + LOG.debug("map=" + map); } - public T create(ChronicleMapBuilder builder) throws IllegalArgumentException, + public T create(ChronicleMapBuilder builder) throws IllegalArgumentException, IOException, TimeoutException, InterruptedException { int withChannelId = nextFreeChannel.incrementAndGet(); - T result = (T) toReplicatedViaChannel(builder, withChannelId).create(); ChronicleMapBuilderWithChannelId builderWithChannelId = new ChronicleMapBuilderWithChannelId(); - builderWithChannelId.channelId = withChannelId; - builderWithChannelId.chronicleMapBuilder = builder; + builderWithChannelId.chronicleMapBuilder = (ChronicleHashBuilder)builder; + map.put(builder.name(), builderWithChannelId); - return result; + + return (T) get(builder.name()).chronicleMapBuilder.create(); + } @@ -188,7 +192,7 @@ class ReplicationHubFindByName implements FindByName { return builder.instance().replicatedViaChannel(replicationHub.createChannel((short) channelId)); } - private ChronicleHashInstanceConfig toReplicatedViaChannel(ChronicleMapBuilder builder, int + private ChronicleHashInstanceConfig toReplicatedViaChannel(ChronicleHashBuilder builder, int withChannelId) { return builder.instance().replicatedViaChannel(replicationHub.createChannel((short) withChannelId)); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 6e55c113e..2510ff683 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -496,6 +496,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // methods throw new IllegalStateException("dropping connection, " + "as the remote-identifier is already being used, identifier=" + remoteIdentifier); + if (LOG.isDebugEnabled()) + LOG.debug("handshaking for localIdentifier=" + localIdentifier + "," + + "remoteIdentifier=" + remoteIdentifier); + attached.remoteModificationIterator = replica.acquireModificationIterator(remoteIdentifier, attached); diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index aaa85fdcf..ef6781247 100755 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -45,8 +45,8 @@ 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 ChronicleHashBuilder} interface - * documentation. + *

    {@code ChronicleMapOnHeapUpdatableBuilder} is mutable, see a note in {@link ChronicleHashBuilder} + * interface documentation. * * @param element type of the sets, created by this builder * @see ChronicleSet @@ -309,6 +309,15 @@ public class ChronicleSetBuilder return new SetFromMap(map); } + @Override + public String name() { + return chronicleMapBuilderOnHeap.name(); + } + + public ChronicleSetBuilder name(String name) { + chronicleMapBuilderOnHeap.name(name); + return this; + } } diff --git a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java index 313b8d1ab..cd27d71ec 100644 --- a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java +++ b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java @@ -33,7 +33,7 @@ import static junit.framework.Assert.assertEquals; public class MapByNameTest { - private ReplicationHubFindByName findMapByName; + private ReplicationHubFindByName findMapByName; @Before public void setUp() throws IOException { @@ -80,10 +80,10 @@ public class MapByNameTest { .name("myMap") .removeReturnsNull(true); - ReplicationHubFindByName mapByName = nodeDiscovery.mapByName(); + ReplicationHubFindByName mapByName = nodeDiscovery.mapByName(); // mapByName.add(builder); - ChronicleMap myMap2 = mapByName.from("myMap"); + ChronicleMap myMap2 = mapByName.from("myMap"); } @@ -91,19 +91,20 @@ public class MapByNameTest { NodeDiscovery nodeDiscovery = new NodeDiscovery(); - final ReplicationHubFindByName mapByName = nodeDiscovery.mapByName(); + final ReplicationHubFindByName mapByName = nodeDiscovery.mapByName(); Thread.sleep(2000); - ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence +/* ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence .class, CharSequence.class) .minSegments(2) + .entries(2) .name("myMap5") .removeReturnsNull(true); - ChronicleMap map = mapByName.create(builder); - map.put("hello", "world"); + ChronicleMap map = mapByName.create(builder); + map.put("hello", "world");*/ - // final ChronicleMap map = mapByName.from("myMap4"); + final ChronicleMap map = mapByName.from("myMap5"); // allow time for replication diff --git a/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java b/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java deleted file mode 100644 index 0189fa77b..000000000 --- a/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java +++ /dev/null @@ -1,165 +0,0 @@ -package net.openhft.chronicle.map.example; - -import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; -import net.openhft.chronicle.map.ChronicleMap; -import net.openhft.chronicle.map.ChronicleMapBuilder; -import net.openhft.chronicle.map.ChronicleMapBuilder; -import org.junit.Assert; -import org.junit.Test; - -import java.io.IOException; -import java.net.InetSocketAddress; - -import static java.util.concurrent.TimeUnit.SECONDS; -import static org.junit.Assert.assertEquals; - -/** - * @author Rob Austin. - */ -public class ChannelExample { - - - - -/* - * Copyright 2014 Higher Frequency Trading - * - * http://www.higherfrequencytrading.com - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - - - /** - * Test ReplicatedChronicleMap where the Replicated is over a TCP Socket, but with 4 nodes - * - * @author Rob Austin. - */ - public class ChannelReplicationTest { - - private ChronicleMap favoriteComputerServer1; - private ChronicleMap favoriteComputerServer2; - - private ChronicleMap favoriteColourServer2; - private ChronicleMap favoriteColourServer1; - - - private ReplicationHub hubOnServer1; - private ReplicationHub hubOnServer2; - - - @Test - public void test() throws IOException, InterruptedException { - - // server 1 with identifier = 1 - { - ChronicleMapBuilder smallStringToStringMapBuilder = - ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(1000); - - byte identifier = (byte) 1; - - TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig - .of(8086, new InetSocketAddress("localhost", 8087)) - .heartBeatInterval(1, SECONDS); - - - hubOnServer1 = ReplicationHub.builder() - .tcpTransportAndNetwork(tcpConfig) - .createWithId(identifier); - - // this demotes favoriteColour - short channel1 = (short) 1; - - favoriteColourServer1 = smallStringToStringMapBuilder.instance() - .replicatedViaChannel(hubOnServer1.createChannel(channel1)).create(); - - favoriteColourServer1.put("peter", "green"); - - // this demotes favoriteComputer - short channel2 = (short) 2; - - favoriteComputerServer1 = smallStringToStringMapBuilder.instance() - .replicatedViaChannel(hubOnServer1.createChannel(channel2)).create(); - - favoriteComputerServer1.put("peter", "dell"); - - } - - // server 2 with identifier = 2 - { - ChronicleMapBuilder smallStringToStringMapBuilder = - ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(1000); - - byte identifier = (byte) 2; - - TcpTransportAndNetworkConfig tcpConfig = - TcpTransportAndNetworkConfig.of(8087).heartBeatInterval(1, SECONDS); - - hubOnServer2 = ReplicationHub.builder() - .tcpTransportAndNetwork(tcpConfig) - .createWithId(identifier); - - - // this demotes favoriteColour - short channel1 = (short) 1; - - favoriteColourServer2 = smallStringToStringMapBuilder.instance() - .replicatedViaChannel(hubOnServer2.createChannel(channel1)).create(); - - - favoriteColourServer2.put("rob", "blue"); - - // this demotes favoriteComputer - short channel2 = (short) 2; - - favoriteComputerServer2 = smallStringToStringMapBuilder.instance() - .replicatedViaChannel(hubOnServer2.createChannel(channel2)).create(); - - favoriteComputerServer2.put("rob", "mac"); - favoriteComputerServer2.put("daniel", "mac"); - } - - - // allow time for the recompilation to resolve - for (int t = 0; t < 2500; t++) { - if (favoriteComputerServer2.equals(favoriteComputerServer1) && - favoriteColourServer2.equals(favoriteColourServer1)) - break; - Thread.sleep(1); - } - - - assertEquals(favoriteComputerServer1, favoriteComputerServer2); - Assert.assertEquals(3, favoriteComputerServer2.size()); - - - assertEquals(favoriteColourServer1, favoriteColourServer2); - Assert.assertEquals(2, favoriteColourServer1.size()); - - - favoriteColourServer1.close(); - favoriteComputerServer2.close(); - favoriteColourServer2.close(); - favoriteComputerServer1.close(); -} - - - } - -} - - - diff --git a/src/test/java/net/openhft/chronicle/map/example/ChannelReplicationTest.java b/src/test/java/net/openhft/chronicle/map/example/ChannelReplicationTest.java new file mode 100644 index 000000000..c185ee069 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/example/ChannelReplicationTest.java @@ -0,0 +1,375 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map.example; + +import net.openhft.chronicle.hash.ChronicleHashInstanceConfig; +import net.openhft.chronicle.hash.replication.ReplicationHub; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.map.ChronicleMap; +import net.openhft.chronicle.map.ChronicleMapBuilder; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.net.InetSocketAddress; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertEquals; + +/** + * @author Rob Austin. + */ + +public class ChannelReplicationTest { + + private ChronicleMap favoriteComputerServer1; + private ChronicleMap favoriteComputerServer2; + + private ChronicleMap favoriteColourServer2; + private ChronicleMap favoriteColourServer1; + + + private ReplicationHub hubOnServer1; + private ReplicationHub hubOnServer2; + + + @Test + public void test() throws IOException, InterruptedException { + + // server 1 with identifier = 1 + { + ChronicleMapBuilder smallStringToStringMapBuilder = + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000); + + byte identifier = (byte) 1; + + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .of(8086, new InetSocketAddress("localhost", 8087)) + .heartBeatInterval(1, SECONDS); + + + hubOnServer1 = ReplicationHub.builder() + .tcpTransportAndNetwork(tcpConfig) + .createWithId(identifier); + + // this demotes favoriteColour + short channel1 = (short) 1; + + favoriteColourServer1 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer1.createChannel(channel1)).create(); + + favoriteColourServer1.put("peter", "green"); + + // this demotes favoriteComputer + short channel2 = (short) 2; + + favoriteComputerServer1 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer1.createChannel(channel2)).create(); + + favoriteComputerServer1.put("peter", "dell"); + + } + + // server 2 with identifier = 2 + { + ChronicleMapBuilder smallStringToStringMapBuilder = + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000); + + byte identifier = (byte) 2; + + TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.of(8087).heartBeatInterval(1, SECONDS); + + hubOnServer2 = ReplicationHub.builder() + .tcpTransportAndNetwork(tcpConfig) + .createWithId(identifier); + + + // this demotes favoriteColour + short channel1 = (short) 1; + + favoriteColourServer2 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer2.createChannel(channel1)).create(); + + + favoriteColourServer2.put("rob", "blue"); + + // this demotes favoriteComputer + short channel2 = (short) 2; + + favoriteComputerServer2 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer2.createChannel(channel2)).create(); + + favoriteComputerServer2.put("rob", "mac"); + favoriteComputerServer2.put("daniel", "mac"); + } + + + // allow time for the recompilation to resolve + for (int t = 0; t < 2500; t++) { + if (favoriteComputerServer2.equals(favoriteComputerServer1) && + favoriteColourServer2.equals(favoriteColourServer1)) + break; + Thread.sleep(1); + } + + + assertEquals(favoriteComputerServer1, favoriteComputerServer2); + Assert.assertEquals(3, favoriteComputerServer2.size()); + + + assertEquals(favoriteColourServer1, favoriteColourServer2); + Assert.assertEquals(2, favoriteColourServer1.size()); + + + favoriteColourServer1.close(); + favoriteComputerServer2.close(); + favoriteColourServer2.close(); + favoriteComputerServer1.close(); + } + + + @Test + public void testPublishOnOneMapOnlyBootstrapTwice() throws IOException, InterruptedException { + + + ChronicleMap favoriteComputerServer3; + + ChronicleMap favoriteColourServer3; + + + // server 1 with identifier = 1 + { + ChronicleMapBuilder smallStringToStringMapBuilder = + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000); + + byte identifier = (byte) 1; + + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .of(8086) + .heartBeatInterval(1, SECONDS); + + + hubOnServer1 = ReplicationHub.builder() + .tcpTransportAndNetwork(tcpConfig) + .createWithId(identifier); + + // this demotes favoriteColour + short channel1 = (short) 1; + + favoriteColourServer1 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer1.createChannel(channel1)).create(); + + favoriteColourServer1.put("peter", "green"); + + // this demotes favoriteComputer + short channel2 = (short) 2; + + favoriteComputerServer1 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer1.createChannel(channel2)).create(); + + favoriteComputerServer1.put("peter", "dell"); + + + } + + // server 2 with identifier = 2 + { + ChronicleMapBuilder builder = + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000); + + byte identifier = (byte) 6; + + TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.of(8087, new InetSocketAddress("localhost", 8086)).heartBeatInterval(1, SECONDS); + + ReplicationHub hubOnServer = ReplicationHub.builder() + .tcpTransportAndNetwork(tcpConfig) + .createWithId(identifier); + + // this demotes favoriteColour + short channel1 = (short) 1; + + ChronicleHashInstanceConfig> instance = builder.instance().replicatedViaChannel(hubOnServer.createChannel(channel1)); + + favoriteColourServer2 = instance.create(); + } + + + { + ChronicleMapBuilder smallStringToStringMapBuilder = + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000); + + byte identifier = (byte) 3; + + TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.of(8088, new InetSocketAddress("localhost", 8086)).heartBeatInterval(1, SECONDS); + + ReplicationHub hubOnServer3 = ReplicationHub.builder() + .tcpTransportAndNetwork(tcpConfig) + .createWithId(identifier); + + + // this demotes favoriteColour + short channel1 = (short) 1; + + favoriteColourServer3 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer3.createChannel(channel1)).create(); + + // this demotes favoriteComputer + short channel2 = (short) 2; + + favoriteComputerServer3 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer3.createChannel(channel2)).create(); + + + } + + + // allow time for the recompilation to resolve + for (int t = 0; t < 250000; t++) { + if (favoriteComputerServer1.equals(favoriteComputerServer3) && + favoriteColourServer1.equals(favoriteColourServer3) && + favoriteColourServer2.equals(favoriteColourServer3)) + break; + Thread.sleep(1); + } + + + assertEquals(favoriteComputerServer3, favoriteComputerServer1); + assertEquals(favoriteColourServer2, favoriteColourServer3); + + + Assert.assertEquals(1, favoriteColourServer3.size()); + Assert.assertEquals(1, favoriteComputerServer1.size()); + + favoriteComputerServer3.close(); + favoriteColourServer3.close(); + + favoriteComputerServer1.close(); + favoriteColourServer1.close(); + + + favoriteColourServer2.close(); + + + } + + + @Test + public void testPublishOnOneMapOnly() throws IOException, InterruptedException { + + // server 1 with identifier = 1 + { + ChronicleMapBuilder smallStringToStringMapBuilder = + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000); + + byte identifier = (byte) 1; + + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .of(8086, new InetSocketAddress("localhost", 8087)) + .heartBeatInterval(1, SECONDS); + + + hubOnServer1 = ReplicationHub.builder() + .tcpTransportAndNetwork(tcpConfig) + .createWithId(identifier); + + // this demotes favoriteColour + short channel1 = (short) 1; + + favoriteColourServer1 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer1.createChannel(channel1)).create(); + + favoriteColourServer1.put("peter", "green"); + + // this demotes favoriteComputer + short channel2 = (short) 2; + + favoriteComputerServer1 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer1.createChannel(channel2)).create(); + + favoriteComputerServer1.put("peter", "dell"); + + } + + // server 2 with identifier = 2 + { + ChronicleMapBuilder smallStringToStringMapBuilder = + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000); + + byte identifier = (byte) 2; + + TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.of(8087).heartBeatInterval(1, SECONDS); + + hubOnServer2 = ReplicationHub.builder() + .tcpTransportAndNetwork(tcpConfig) + .createWithId(identifier); + + + // this demotes favoriteColour + short channel1 = (short) 1; + + favoriteColourServer2 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer2.createChannel(channel1)).create(); + + + // this demotes favoriteComputer + short channel2 = (short) 2; + + favoriteComputerServer2 = smallStringToStringMapBuilder.instance() + .replicatedViaChannel(hubOnServer2.createChannel(channel2)).create(); + } + + + // allow time for the recompilation to resolve + for (int t = 0; t < 2500; t++) { + if (favoriteComputerServer2.equals(favoriteComputerServer1) && + favoriteColourServer2.equals(favoriteColourServer1)) + break; + Thread.sleep(1); + } + + + assertEquals(favoriteComputerServer1, favoriteComputerServer2); + Assert.assertEquals(1, favoriteComputerServer2.size()); + + + assertEquals(favoriteColourServer1, favoriteColourServer2); + Assert.assertEquals(1, favoriteColourServer1.size()); + + + favoriteColourServer1.close(); + favoriteComputerServer2.close(); + favoriteColourServer2.close(); + favoriteComputerServer1.close(); + } +} + + + +