From 9dec1ed4db84024ee8c979073bee6dacbc852c43 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 4 Nov 2014 14:44:59 +0000 Subject: [PATCH] reverted cff7c5bea8328a2f11a201f0f32f45edc5a8f9ee --- .../chronicle/hash/ChronicleHashBuilder.java | 152 +++++++----- .../hash/serialization/BytesInterop.java | 23 +- .../hash/serialization/BytesReader.java | 75 +----- .../hash/serialization/BytesWriter.java | 53 ----- .../impl/ByteArrayMarshaller.java | 10 +- .../impl/ByteableMarshaller.java | 17 +- .../impl/CharArrayMarshaller.java | 10 +- .../impl/CharSequenceReader.java | 18 +- .../serialization/impl/DoubleMarshaller.java | 8 +- .../serialization/impl/IntegerMarshaller.java | 8 +- .../serialization/impl/LongMarshaller.java | 9 +- .../serialization/impl/VoidMarshaller.java | 9 +- .../map/AbstractChannelReplicator.java | 2 +- .../map/AbstractChronicleMapBuilder.java | 224 +++++++++--------- .../chronicle/map/ChronicleMapBuilder.java | 10 +- .../openhft/chronicle/map/NodeDiscovery.java | 37 ++- .../OffHeapUpdatableChronicleMapBuilder.java | 2 - .../net/openhft/chronicle/map/Replica.java | 2 +- .../chronicle/map/ReplicatedChronicleMap.java | 11 +- .../net/openhft/chronicle/map/Replicator.java | 2 +- .../openhft/chronicle/map/Replicators.java | 23 +- .../chronicle/map/SerializationBuilder.java | 46 ++-- .../chronicle/map/StatelessChronicleMap.java | 17 +- .../openhft/chronicle/map/TcpReplicator.java | 76 ++---- .../chronicle/map/UdpChannelReplicator.java | 4 +- .../openhft/chronicle/map/UdpReplicator.java | 6 +- .../chronicle/map/VanillaChronicleMap.java | 28 ++- .../openhft/chronicle/map/WriteContext.java | 3 +- .../chronicle/map/ZeroOutValueBytes.java | 2 +- .../chronicle/set/ChronicleSetBuilder.java | 50 ++-- .../chronicle/set/DummyValueMarshaller.java | 9 +- src/test/java/eg/BigData.java | 2 +- .../chronicle/map/AcquireGetUsingMain.java | 4 +- .../net/openhft/chronicle/map/Builder.java | 7 +- .../chronicle/map/CHMLatencyTestMain.java | 2 +- .../net/openhft/chronicle/map/CHMTest5.java | 3 +- .../chronicle/map/ChannelReplicationTest.java | 37 +-- .../chronicle/map/ChronicleMapTest.java | 103 ++------ .../chronicle/map/DefaultValueTest.java | 6 +- .../chronicle/map/EntryCountMapTest.java | 5 +- .../openhft/chronicle/map/OSResizesMain.java | 8 +- .../map/PostChannelCreationTest.java | 35 +-- .../openhft/chronicle/map/SerializerTest.java | 4 +- .../chronicle/map/SingleMapChannelTest.java | 34 +-- .../chronicle/map/StatelessClientTest.java | 68 +++--- .../map/TCPSocketReplication4WayMapTest.java | 12 + .../TCPSocketReplicationBootStrapTests.java | 18 +- .../map/TCPSocketReplicationIntValueTest.java | 12 +- .../map/TCPSocketReplicationTest.java | 3 +- ...cketReplicationTest3wayPutReturnsNull.java | 7 +- ...CPSocketReplicationTestPostConnection.java | 6 +- .../map/TimeBasedReplicationTests.java | 90 ++++--- .../map/TwoMapOnDifferentServers.java | 12 +- .../map/UDPSocketReplicationTest.java | 8 +- .../chronicle/map/example/ChannelExample.java | 59 +++-- .../map/example/DistributedSequenceMain.java | 3 + .../OpenJDKAndHashMapExamplesTest.java | 6 +- .../pingpong_latency/PingPongCASLeft.java | 7 + .../map/ipc/StateMachineTutorial.java | 5 + .../map/jrs166/map/ChronicleMapTest.java | 24 ++ .../chronicle/map/jrs166/map/MapCheck.java | 3 + .../jrs166/map/StatelessChronicleMapTest.java | 14 +- .../map/utility/ProcessInstanceLimiter.java | 3 +- .../SetForTCPSocketReplicationTest3way.java | 9 +- 64 files changed, 700 insertions(+), 865 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index 4f86d0980..ab5624b35 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -18,9 +18,6 @@ package net.openhft.chronicle.hash; -import net.openhft.chronicle.hash.replication.SimpleReplication; -import net.openhft.chronicle.hash.replication.TcpConfig; -import net.openhft.chronicle.hash.replication.TimeProvider; import net.openhft.chronicle.map.*; import net.openhft.chronicle.set.ChronicleSet; import net.openhft.chronicle.set.ChronicleSetBuilder; @@ -35,7 +32,6 @@ import org.jetbrains.annotations.NotNull; import java.io.Externalizable; import java.io.File; import java.io.IOException; -import java.net.InetSocketAddress; import java.util.HashMap; import java.util.HashSet; import java.util.Map; @@ -228,6 +224,67 @@ public interface ChronicleHashBuilder + * the section about replication in ChronicleMap manual for more information. + * + *

Another way to establish replication is {@link #channel(ChannelProvider.ChronicleChannel)} + * method. + * + *

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 #channel(ChannelProvider.ChronicleChannel)} method. + * + * @param identifier the new {@linkplain Replica#identifier() identifier} of the containers, + * created by this builder + * @param replicators several replicators, used to sync the data across replicating nodes + * independently + * @return this builder back + * @see #channel(ChannelProvider.ChronicleChannel) + * @see #disableReplication() + */ + B replicators(byte identifier, ReplicationConfig... replicators); + + /** + * Configures replication of the hash containers, created by this builder, via so called + * "channels". See the + * section about Channels and ChannelProvider in ChronicleMap manual for more information. + * + *

Another way to establish replication if {@link #replicators(byte, ReplicationConfig[])} + * method. + * + *

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 means of this method or {@link #replicators(byte, ReplicationConfig[])} method + * calls. + * + * @param chronicleChannel the channel responsible for gathering updates of hash containers, + * created by this builder, and replicating them over network + * @return this builder object back + * @see #replicators(byte, ReplicationConfig[]) + * @see #disableReplication() + */ + B channel(ChannelProvider.ChronicleChannel chronicleChannel); + + /** + * Specifies that hash containers, created by this builder, shouldn't replicate their data over + * network. This method call clears any replication configurations, made via {@link + * #replicators(byte, ReplicationConfig[])} or {@link #channel(ChannelProvider.ChronicleChannel)} + * method call. + * + *

By default replication is not configured, so this method is useful only to ensure + * that hash containers, created by some inherited or passed builder, wouldn't replicate. I. e. + * in rare cases. + * + * @return this builder back + * @see #replicators(byte, ReplicationConfig[]) + * @see #channel(ChannelProvider.ChronicleChannel) + */ + B disableReplication(); + /** * Configures a time provider, used by hash containers, created by this builder, for needs of * replication consensus protocol (conflicting data updates resolution). @@ -236,7 +293,8 @@ public interface ChronicleHashBuilder - * the section about replication in ChronicleMap manual for more information. + * Specifies that hash containers, created by this builder, should be a stateless + * implementation, this stateless implementation will be referred to as a stateless client as it + * will not hold any of its own data locally, the stateless client will perform Remote Procedure + * Calls ( RPC ) to another {@link ChronicleMap} or {@link ChronicleSet} which we will refer to + * as the server. The server will hold all your data, the server can not it’s self be a + * stateless client. Your stateless client must be connected to the server via TCP/IP. The + * stateless client will delegate all your method calls to the remote sever. The stateless + * client operations will block, in other words the stateless client will wait for the server to + * send a response before continuing to the next operation. The stateless client could be + * consider to be a ClientProxy to {@link ChronicleMap} or {@link ChronicleSet} running on + * another host * - *

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, TcpConfig)} shortcut method. - * - * @param replication the replication config + * @param statelessBuilder Stateless Client Config, {@code null} demotes no statelessBuilder * @return this builder back - * @see ChronicleHashInstanceConfig#replicated(SimpleReplication) - * @see #replication(byte, TcpConfig) */ - B replication(SimpleReplication replication); + B stateless(StatelessBuilder statelessBuilder); + /** - * Shortcut for {@code replication(SimpleReplication.builder() - * .tcpTransportAndNetwork(tcpTransportAndNetwork).create(identifier))}. + * Specifies that hash containers us the specified file, if create is called and this is not set + * the map will be created in off-heap memory, and changes will not be persisted to disk. If a + * file is provided all changes to the map are persisted to disk (this is an operating system + * guarantee) independently from JVM process lifecycle. * - * @param identifier the network-wide identifier of the containers, created by this builder - * @param tcpTransportAndNetwork configuration of tcp connection and network + * @param file the file with existing hash container or a desired location of a new off-heap + * persisted hash container * @return this builder back - * @see #replication(SimpleReplication) - * @see ChronicleHashInstanceConfig#replicated(byte, TcpConfig) + * @throws IOException If the file cannot be created and read. + * @see ChronicleHash#file() + * @see ChronicleHash#close() */ - B replication(byte identifier, TcpConfig tcpTransportAndNetwork); - - StatelessClientConfig statelessClient(InetSocketAddress remoteAddress); - - ChronicleHashInstanceConfig instance(); + B file(File file) throws IOException; /** * Creates a new hash container, storing it's data in off-heap memory, not mapped to any file. @@ -367,42 +426,15 @@ public interface ChronicleHashBuilderThis 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 occurs - * @see #createPersistedTo(File) - * @see #instance() + * @see #create() */ 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. - * - *

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

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 - * @return a hash container mapped to the given file - * @throws IOException if any IO error, related to off-heap memory allocation or file mapping - * occurs - * @see ChronicleHash#file() - * @see ChronicleHash#close() - * @see #create() - * @see ChronicleHashInstanceConfig#persistedTo(File) - */ - C createPersistedTo(File file) throws IOException; + + + + } diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/BytesInterop.java b/src/main/java/net/openhft/chronicle/hash/serialization/BytesInterop.java index 204a49f69..33231635c 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/BytesInterop.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/BytesInterop.java @@ -22,30 +22,15 @@ import net.openhft.lang.io.Bytes; import net.openhft.lang.model.Byteable; /** - * Writer for objects, which themselves are bytes sequence in some sense: - * {@link Byteable Byteables}, {@code byte[]} arrays, other primitive arrays, "flat" objects - * (with primitives fields only), particularly boxed primitive types. - * - *

This interface is called "BytesInterop" because it allows to work with object as they - * are already marshalled to {@code Bytes}: {@linkplain #startsWith(Bytes, Object) compare} - * with other {@code Bytes}, i. e. interoperate objects and {@code Bytes}. + * Marshaller for byte sequences, which are copied to off-heap {@link Bytes} + * in a very straightforward manner, e. g. {@link Byteable Byteables}, {@code byte[]} arrays, + * {@code Bytes} themselves. The criterion of this interface applicability -- + * {@link Object#equals(Object)} implementation shouldn't require deserialization and any garbage creation. * * @param type of marshalled objects */ public interface BytesInterop extends BytesWriter { - /** - * Checks if the given {@code bytes} starts (from the {@code bytes}' {@linkplain - * Bytes#position() position}) with the byte sequence the given object - * {@linkplain #write(Bytes, Object) is serialized} to, without actual serialization. - * - * @param bytes the bytes to check if starts with the serialized form of the given object. - * {@code bytes}' is positioned at the first byte to compare. {@code bytes}' limit - * is unspecified. {@code bytes}' position and limit shouldn't be altered during - * this call. - * @param e the object to serialize virtually and compare with the given {@code bytes} - * @return if the given {@code bytes} starts with the given {@code e} object's serialized form - */ boolean startsWith(Bytes bytes, E e); long hash(E e); diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/BytesReader.java b/src/main/java/net/openhft/chronicle/hash/serialization/BytesReader.java index b43ba57aa..d2d62e49e 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/BytesReader.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/BytesReader.java @@ -18,86 +18,13 @@ package net.openhft.chronicle.hash.serialization; -import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.lang.io.Bytes; -import org.jetbrains.annotations.Nullable; import java.io.Serializable; -/** - * Deserializer (object from {@link Bytes}, mirroring the {@link BytesWriter}, i. e. assuming the - * length of the serialized form isn't written in the beginning of the serialized form itself, - * but managed by {@link ChronicleHash} implementation and passed to the reading methods. - * - *

Implementation example:


- * class LongPair { long first, second; }
- * 
- * enum LongPairArrayReader implements BytesReader<LongPair[]> {
- *     INSTANCE;
- *
- *     @Override
- *     public LongPair[] read(Bytes bytes, long size) {
- *         return read(bytes, size, null);
- *     }
- *
- *     @Override
- *     public LongPair[] read(Bytes bytes, long size, LongPair[] toReuse) {
- *         int resLen = (int) (size / 16L);
- *         LongPair[] res;
- *         if (toReuse != null) {
- *             if (toReuse.length == resLen) {
- *                 res = toReuse;
- *             } else {
- *                 res = Arrays.copyOf(toReuse, resLen);
- *             }
- *         } else {
- *             res = new LongPair[resLen];
- *         }
- *         for (int i = 0; i < resLen; i++) {
- *             LongPair pair = res[i];
- *             if (pair == null)
- *                 res[i] = pair = new LongPair();
- *             pair.first = bytes.readLong();
- *             pair.second = bytes.readLong();
- *         }
- *         return res;
- *     }
- * }
- * - * @param the type of the object demarshalled - * @see BytesWriter - */ public interface BytesReader extends Serializable { - /** - * Reads and returns the object from {@code [position(), position() + size]} bytes of the given - * {@link Bytes}. - * - *

Implementation of this method should increment the {@code bytes}' position by the given - * size, i. e. "consume" the bytes of the deserialized object. It must not alter the - * {@code bytes}' {@link Bytes#limit() limit} and contents. - * - * @param bytes the {@code Bytes} to read the object from - * @param size the size of the serialized form of the returned object - * @return the object read from the bytes - * @see BytesWriter#write(Bytes, Object) - * @see #read(Bytes, long, Object) - */ E read(Bytes bytes, long size); - /** - * Similar to {@link #read(Bytes, long)}, but should attempt to reuse the given object, i. e. - * to read the deserialized data into the given instance. If it is possible, this objects than - * returned from this method. If it isn't possible for any reason, a new object should be - * created and returned, just like in {@link #read(Bytes, long)} method. The given object could - * be {@code null}, in this case this method should behave exactly the same as - * {@link #read(Bytes, long)} method does. - * - * @param bytes the {@code Bytes} to read the object from - * @param size the size of the serialized form of the returned object - * @param toReuse the object to read the deserialized data into - * @return the object read from the bytes, either reused or newly created - * @see #read(Bytes, long) - */ - E read(Bytes bytes, long size, @Nullable E toReuse); + E read(Bytes bytes, long size, E e); } diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/BytesWriter.java b/src/main/java/net/openhft/chronicle/hash/serialization/BytesWriter.java index 957c2a3dc..f14c31fa2 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/BytesWriter.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/BytesWriter.java @@ -18,66 +18,13 @@ package net.openhft.chronicle.hash.serialization; -import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.lang.io.Bytes; import java.io.Serializable; -/** - * Serializer (object to {@link Bytes} writer) which 1) knows the length of serialized form - * of any object before actual serialization 2) doesn't include that length in the serialized - * form itself, assuming it will be passed by the {@link ChronicleHash} into - * {@link BytesReader#read(Bytes, long)} deserialization method. - * - *

Implementation example: 


- * class LongPair { long first, second; }
- *
- * enum LongPairArrayWriter implements BytesWriter<LongPair[]> {
- *     INSTANCE;
- *
- *     @Override
- *     public long size(LongPair[] longPairs) {
- *         return longPairs.length * 16L;
- *     }
- *
- *     @Override
- *     public void write(Bytes bytes, LongPair[] longPairs) {
- *         for (LongPair pair : longPairs) {
- *             bytes.writeLong(pair.first);
- *             bytes.writeLong(pair.second);
- *         }
- *     }
- * }
- * - * @param the type of the object marshalled - * @see BytesReader - * @see BytesInterop - */ public interface BytesWriter extends Serializable { - /** - * Returns the length (in bytes) of the serialized form of the given object. Serialization form - * in terms of this interface, i. e. how much bytes are written to {@code bytes} on - * {@link #write(Bytes, Object) write(bytes, e)} call. - * - * @param e the object which serialized form length should be returned - * @return the length (in bytes) of the serialized form of the given object - */ long size(E e); - /** - * Serializes the given object to the given {@code bytes}, without writing the length of the - * serialized form itself. - * - *

Implementation of this method should increment the {@code bytes}' {@linkplain - * Bytes#position() position} by {@link #size(Object) size(e)}. The given object should - * be written into these range between the initial {@code bytes}' position and the position - * after this method call returns. Implementation of this method must not alter the - * {@code bytes}' {@linkplain Bytes#limit() limit} and contents outside of the specified range. - * - * @param bytes the {@code Bytes} to write the given object to - * @param e the object to serialize to the given bytes - * @see BytesReader#read(Bytes, long) - */ void write(Bytes bytes, E e); } diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/impl/ByteArrayMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/impl/ByteArrayMarshaller.java index 43f11cdc5..8ea01472e 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/impl/ByteArrayMarshaller.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/impl/ByteArrayMarshaller.java @@ -59,10 +59,10 @@ public enum ByteArrayMarshaller implements BytesInterop, BytesReader - implements BytesInterop, BytesReader, SizeMarshaller { +public class ByteableMarshaller implements AgileBytesMarshaller { private static final long serialVersionUID = 0L; public static ByteableMarshaller of(@NotNull Class eClass) { @@ -117,13 +114,13 @@ public class ByteableMarshaller } @Override - public E read(Bytes bytes, long size, E toReuse) { + public E read(Bytes bytes, long size, E e) { try { - if (toReuse == null) - toReuse = getInstance(); - setBytesAndOffset(toReuse, bytes); + if (e == null) + e = getInstance(); + setBytesAndOffset(e, bytes); bytes.skip(size); - return toReuse; + return e; } catch (Exception ex) { throw new IllegalStateException(ex); } diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/impl/CharArrayMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/impl/CharArrayMarshaller.java index 1f285d8ae..1cc1da171 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/impl/CharArrayMarshaller.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/impl/CharArrayMarshaller.java @@ -49,11 +49,11 @@ public enum CharArrayMarshaller implements BytesInterop, BytesReader } @Override - public S read(Bytes bytes, long size, S toReuse) { + public S read(Bytes bytes, long size, S s) { Appendable appendable; - if (toReuse instanceof Appendable) { - appendable = (Appendable) toReuse; - if (toReuse instanceof StringBuilder) { - ((StringBuilder) toReuse).setLength(0); - } else if (toReuse instanceof StringBuffer) { - ((StringBuffer) toReuse).setLength(0); + if (s instanceof Appendable) { + appendable = (Appendable) s; + if (s instanceof StringBuilder) { + ((StringBuilder) s).setLength(0); + } else if (s instanceof StringBuffer) { + ((StringBuffer) s).setLength(0); } } else { sb.setLength(0); @@ -96,8 +96,8 @@ public final class CharSequenceReader } catch (IOException e) { throw new IllegalStateException(e); } - if (appendable == toReuse) - return toReuse; + if (appendable == s) + return s; return interner.intern(sb); } diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/impl/DoubleMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/impl/DoubleMarshaller.java index 92a99b534..1f663ca21 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/impl/DoubleMarshaller.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/impl/DoubleMarshaller.java @@ -18,15 +18,13 @@ package net.openhft.chronicle.hash.serialization.impl; -import net.openhft.chronicle.hash.serialization.BytesInterop; -import net.openhft.chronicle.hash.serialization.BytesReader; +import net.openhft.chronicle.hash.serialization.AgileBytesMarshaller; import net.openhft.chronicle.hash.serialization.Hasher; -import net.openhft.chronicle.hash.serialization.SizeMarshaller; import net.openhft.lang.io.Bytes; import static java.lang.Double.doubleToLongBits; -public enum DoubleMarshaller implements BytesInterop, BytesReader, SizeMarshaller { +public enum DoubleMarshaller implements AgileBytesMarshaller { INSTANCE; @Override @@ -70,7 +68,7 @@ public enum DoubleMarshaller implements BytesInterop, BytesReader, BytesReader, SizeMarshaller { +public enum IntegerMarshaller implements AgileBytesMarshaller { INSTANCE; @Override @@ -66,7 +66,7 @@ public enum IntegerMarshaller } @Override - public Integer read(Bytes bytes, long size, Integer toReuse) { + public Integer read(Bytes bytes, long size, Integer e) { return bytes.readInt(); } } diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/impl/LongMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/impl/LongMarshaller.java index 1e4f8db01..822d063a7 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/impl/LongMarshaller.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/impl/LongMarshaller.java @@ -18,14 +18,11 @@ package net.openhft.chronicle.hash.serialization.impl; -import net.openhft.chronicle.hash.serialization.BytesInterop; -import net.openhft.chronicle.hash.serialization.BytesReader; +import net.openhft.chronicle.hash.serialization.AgileBytesMarshaller; import net.openhft.chronicle.hash.serialization.Hasher; -import net.openhft.chronicle.hash.serialization.SizeMarshaller; import net.openhft.lang.io.Bytes; -public enum LongMarshaller - implements BytesInterop, BytesReader, SizeMarshaller { +public enum LongMarshaller implements AgileBytesMarshaller { INSTANCE; @Override @@ -69,7 +66,7 @@ public enum LongMarshaller } @Override - public Long read(Bytes bytes, long size, Long toReuse) { + public Long read(Bytes bytes, long size, Long e) { return bytes.readLong(); } } diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/impl/VoidMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/impl/VoidMarshaller.java index e743414e7..81f2a5ec5 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/impl/VoidMarshaller.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/impl/VoidMarshaller.java @@ -18,13 +18,10 @@ package net.openhft.chronicle.hash.serialization.impl; -import net.openhft.chronicle.hash.serialization.BytesInterop; -import net.openhft.chronicle.hash.serialization.BytesReader; -import net.openhft.chronicle.hash.serialization.SizeMarshaller; +import net.openhft.chronicle.hash.serialization.AgileBytesMarshaller; import net.openhft.lang.io.Bytes; -public enum VoidMarshaller - implements BytesInterop, BytesReader, SizeMarshaller { +public enum VoidMarshaller implements AgileBytesMarshaller { INSTANCE; @Override @@ -69,7 +66,7 @@ public enum VoidMarshaller } @Override - public Void read(Bytes bytes, long size, Void toReuse) { + public Void read(Bytes bytes, long size, Void aVoid) { return null; } diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 3c88d5c7e..e63616205 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.ThrottlingConfig; +import net.openhft.chronicle.hash.ThrottlingConfig; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; import net.openhft.lang.thread.NamedThreadFactory; diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java index 1fd906ff7..ca7d43f3d 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -19,7 +19,6 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.*; -import net.openhft.chronicle.hash.replication.*; import net.openhft.chronicle.hash.serialization.MetaBytesInterop; import net.openhft.chronicle.hash.serialization.MetaBytesWriter; import net.openhft.chronicle.hash.serialization.MetaProvider; @@ -41,11 +40,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.*; -import java.net.InetSocketAddress; import java.nio.ByteBuffer; -import java.util.ArrayList; import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -107,6 +103,8 @@ public abstract class AbstractChronicleMapBuilder keyBuilder; SerializationBuilder valueBuilder; + private Map, Replicator> replicators = new HashMap, Replicator>(); + // used when configuring the number of segments. private int minSegments = -1; private int actualSegments = -1; @@ -135,8 +133,10 @@ public abstract class AbstractChronicleMapBuilder defaultValueProvider = NullValueProvider.INSTANCE; private PrepareValueBytes prepareValueBytes = null; + private byte identifier = -1; - private SimpleReplication simpleReplication = null; + private StatelessBuilder statelessBuilder; + private File file; AbstractChronicleMapBuilder(Class keyClass, Class valueClass) { keyBuilder = new SerializationBuilder(keyClass, SerializationBuilder.Role.KEY); @@ -283,14 +283,14 @@ public abstract class AbstractChronicleMapBuilder 0) return entryAndValueAlignment().alignSize(entrySize); int size = metaDataBytes; int keySize = keySize(); size += keyBuilder.sizeMarshaller().sizeEncodingSize(keySize); size += keySize; - if (replicated) + if (useReplication()) size += ReplicatedChronicleMap.ADDITIONAL_ENTRY_BYTES; int valueSize = valueSize(); size += valueBuilder.sizeMarshaller().sizeEncodingSize(valueSize); @@ -559,6 +559,43 @@ public abstract class AbstractChronicleMapBuilder> statelessClient( - InetSocketAddress remoteAddress) { - return new StatelessMapConfig<>(this.clone(), remoteAddress); + public B stateless(@NotNull StatelessBuilder statelessBuilder) { + this.statelessBuilder = statelessBuilder; + return self(); + } + + @Override + public B file(File file) throws IOException { + this.file = file; + return self(); } /** @@ -705,7 +748,7 @@ public abstract class AbstractChronicleMapBuilder defaultValueProvider() { if (defaultValueProvider != null) @@ -723,43 +766,7 @@ public abstract class AbstractChronicleMapBuilder(defaultValue, metaValueWriter, valueWriter); } - @Override - public B replication(SimpleReplication replication) { - this.simpleReplication = replication; - return self(); - } - - @Override - public B replication(byte identifier, TcpConfig tcpTransportAndNetwork) { - return replication(SimpleReplication.builder() - .tcpTransportAndNetwork(tcpTransportAndNetwork).create(identifier)); - } - - @Override - public ChronicleHashInstanceConfig> instance() { - return new InstanceConfig<>(this.clone(), simpleReplication, null, null); - } - - @Override - public ChronicleMap createPersistedTo(File file) throws IOException { - return createWithFile(file, simpleReplication, null); - } - - @Override - public ChronicleMap create() throws IOException { - return createWithoutFile(simpleReplication, null); - } - - ChronicleMap create(InstanceConfig ib) throws IOException { - if (ib.file != null) { - return createWithFile(ib.file, ib.simpleReplication, ib.channel); - } else { - return createWithoutFile(ib.simpleReplication, ib.channel); - } - } - - ChronicleMap createWithFile(File file, SimpleReplication simpleReplication, - ReplicationChannel channel) throws IOException { + public ChronicleMap createWithFile(File file) throws IOException { for (int i = 0; i < 10; i++) { if (file.exists() && file.length() > 0) { FileInputStream fis = new FileInputStream(file); @@ -769,7 +776,7 @@ public abstract class AbstractChronicleMapBuilder) ois.readObject(); map.headerSize = roundUpMapHeaderSize(fis.getChannel().position()); map.createMappedStoreAndSegments(file); - return establishReplication(map, simpleReplication, channel); + return establishReplication(map); } catch (ClassNotFoundException e) { throw new IOException(e); } finally { @@ -789,7 +796,7 @@ public abstract class AbstractChronicleMapBuilder map = newMap(simpleReplication, channel); + VanillaChronicleMap map = newMap(); FileOutputStream fos = new FileOutputStream(file); ObjectOutputStream oos = new ObjectOutputStream(fos); @@ -802,66 +809,54 @@ public abstract class AbstractChronicleMapBuilder createReplicated(byte identifier) throws IOException { - VanillaChronicleMap map = - new ReplicatedChronicleMap, - V, Object, MetaBytesWriter>(this, identifier); + + @Override + public ChronicleMap create() throws IOException { + + if (statelessBuilder != null) + return createStatelessMap(statelessBuilder); + + return (file != null) ? createWithFile(file) : createWithoutFile(); + } + + private ChronicleMap createWithoutFile() throws IOException { + VanillaChronicleMap map = newMap(); BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE, map.sizeInBytes(), true); map.createMappedStoreAndSegments(bytesStore); - return map; + return establishReplication(map); } - private ChronicleMap createWithoutFile( - SimpleReplication simpleReplication, ReplicationChannel channel) throws IOException { - VanillaChronicleMap map = newMap(simpleReplication, channel); - BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE, - map.sizeInBytes(), true); - map.createMappedStoreAndSegments(bytesStore); - return establishReplication(map, simpleReplication, channel); - } - - ChronicleMap createStatelessMap(StatelessMapConfig statelessBuilder) - throws IOException { - preMapConstruction(false); + private ChronicleMap createStatelessMap(StatelessBuilder statelessBuilder) throws IOException { + preMapConstruction(); return new StatelessChronicleMap( statelessBuilder, this); } - private VanillaChronicleMap newMap( - SimpleReplication simpleReplication, ReplicationChannel channel) throws IOException { - boolean replicated = simpleReplication != null || channel != null; - preMapConstruction(replicated); - if (replicated) { - byte identifier; - if (simpleReplication != null) { - identifier = simpleReplication.identifier(); - } else { - identifier = channel.hub().identifier(); - } + private VanillaChronicleMap newMap() throws IOException { + preMapConstruction(); + if (useReplication()) { return new ReplicatedChronicleMap, - V, Object, MetaBytesWriter>(this, identifier); + V, Object, MetaBytesWriter>(this); } else { return new VanillaChronicleMap, V, Object, MetaBytesWriter>(this); } } - void preMapConstruction(boolean replicated) { + private boolean useReplication() { + return identifier != -1; + } + + void preMapConstruction() { keyBuilder.objectSerializer(objectSerializer()); valueBuilder.objectSerializer(objectSerializer()); - long maxSize = (long) entrySize(replicated) * figureBufferAllocationFactor(); + long maxSize = (long) entrySize() * figureBufferAllocationFactor(); keyBuilder.maxSize(maxSize); valueBuilder.maxSize(maxSize); @@ -871,37 +866,32 @@ public abstract class AbstractChronicleMapBuilder establishReplication( - VanillaChronicleMap map, SimpleReplication simpleReplication, - ReplicationChannel channel) throws IOException { + private ChronicleMap establishReplication(VanillaChronicleMap map) + throws IOException { if (map instanceof ReplicatedChronicleMap) { - if (simpleReplication != null && channel != null) { - throw new AssertionError("Only one non-null replication should be passed"); - } ReplicatedChronicleMap result = (ReplicatedChronicleMap) map; - List replicators = new ArrayList<>(2); - if (simpleReplication != null) { - if (simpleReplication.tcpTransportAndNetwork() != null) - replicators.add(Replicators.tcp(simpleReplication.tcpTransportAndNetwork())); - if (simpleReplication.udpTransport() != null) - replicators.add(Replicators.udp(simpleReplication.udpTransport())); - } else { - ReplicationHub hub = channel.hub(); - int entrySize = entrySize(true); - if (entrySize > hub.maxEntrySize()) { - throw new IllegalArgumentException("During ChannelProviderBuilder setup, " + - "maxEntrySize=" + hub.maxEntrySize() + - " was specified, but map with " + - "entrySize=" + entrySize + " is attempted to apply to the replicator"); + for (Replicator replicator : replicators.values()) { + if (replicator instanceof ChannelProvider.ChronicleChannel) { + ChannelProvider.ChronicleChannel channel = + (ChannelProvider.ChronicleChannel) replicator; + if (entrySize() > channel.provider().maxEntrySize()) { + throw new IllegalArgumentException("During ChannelProviderBuilder setup, " + + "maxEntrySize=" + channel.provider().maxEntrySize() + + " was specified, but map with " + + "entrySize=" + entrySize() + " is attempted to apply" + + "to the replicator"); + } } - ChannelProvider provider = ChannelProvider.getProvider(hub); - ChannelProvider.ChronicleChannel ch = provider.createChannel(channel.channelId()); - replicators.add(ch); - } - for (Replicator replicator : replicators) { Closeable token = replicator.applyTo(this, result, result, map); if (replicators.size() == 1 && token.getClass() == UdpReplicator.class) { - LOG.warn(Replicators.ONLY_UDP_WARN_MESSAGE); + LOG.warn( + "MISSING TCP REPLICATION : The UdpReplicator only attempts to read data " + + "(it does not enforce or guarantee delivery), you should use" + + "the UdpReplicator if you have a large number of nodes, and you wish" + + "to receive the data before it becomes available on TCP/IP. Since data" + + "delivery is not guaranteed, it is recommended that you only use" + + "the UDP Replicator in conjunction with a TCP Replicator" + ); } result.addCloseable(token); } @@ -916,7 +906,11 @@ public abstract class AbstractChronicleMapBuilder return this; } - public ChronicleMapBuilder valueMarshallers( - BytesWriter valueWriter, BytesReader valueReader) { - valueBuilder.writer(valueWriter); - valueBuilder.reader(valueReader, valueBuilder.factory()); + public ChronicleMapBuilder valueMarshaller( + @NotNull AgileBytesMarshaller valueMarshaller) { + valueBuilder.agileMarshaller(valueMarshaller, null); return this; } } diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index 97eca2983..16bb25b95 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -18,9 +18,8 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.IdentifierListener; -import net.openhft.chronicle.hash.replication.TcpConfig; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.TcpReplicationConfig; +import net.openhft.chronicle.hash.UdpReplicationConfig; import net.openhft.lang.collection.ATSDirectBitSet; import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.io.ByteBufferBytes; @@ -42,7 +41,7 @@ import java.util.concurrent.atomic.AtomicReference; import static java.nio.ByteBuffer.allocateDirect; import static java.util.concurrent.TimeUnit.SECONDS; -import static net.openhft.chronicle.hash.replication.UdpConfig.simple; +import static net.openhft.chronicle.hash.UdpReplicationConfig.simple; import static net.openhft.chronicle.map.ConcurrentExpiryMap.getDefaultAddress; import static net.openhft.chronicle.map.DiscoveryNodeBytesMarshallable.ProposedNodes; import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.BOOTSTRAP_BYTES; @@ -56,7 +55,7 @@ import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG; public class NodeDiscovery { private final AddressAndPort ourAddressAndPort; - private final UdpConfig udpConfig; + private final UdpReplicationConfig udpConfig; private final DiscoveryNodeBytesMarshallable discoveryNodeBytesMarshallable; private final AtomicReference nodeDiscoveryEventListenerAtomicReference = new AtomicReference(); @@ -221,11 +220,9 @@ public class NodeDiscovery { // add our identifier and host:port to the list of known identifiers knownNodes.add(ourAddressAndPort, identifier); - final TcpConfig tcpConfig = TcpConfig - .unknownTopology(ourAddressAndPort.getPort(), - toInetSocketCollection(knownHostPorts)) - .heartBeatInterval(1, SECONDS) - .nonUniqueIdentifierListener(identifierListener); + final TcpReplicationConfig tcpConfig = TcpReplicationConfig + .of(ourAddressAndPort.getPort(), toInetSocketArray(knownHostPorts)) + .heartBeatInterval(1, SECONDS).nonUniqueIdentifierListener(identifierListener); LOG.info("Using Remote identifier=" + identifier); @@ -234,26 +231,28 @@ public class NodeDiscovery { return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) - .replication(identifier, tcpConfig).create(); + .replicators(identifier, tcpConfig).create(); } - private static Collection toInetSocketCollection(Set source) - throws UnknownHostException { + private InetSocketAddress[] toInetSocketArray(Set source) throws + UnknownHostException { // make a safe copy final HashSet addressAndPorts = new HashSet(source); if (addressAndPorts.isEmpty()) - return Collections.emptyList(); + return new InetSocketAddress[0]; - ArrayList addresses = - new ArrayList(addressAndPorts.size()); + final InetSocketAddress[] addresses = new InetSocketAddress[addressAndPorts.size()]; + + int i = 0; for (final AddressAndPort addressAndPort : addressAndPorts) { - addresses.add(new InetSocketAddress(InetAddress.getByAddress(addressAndPort.address()) - .getHostAddress(), addressAndPort.port())); + addresses[i++] = new InetSocketAddress(InetAddress.getByAddress(addressAndPort.address()) + .getHostAddress(), addressAndPort.port()); } + return addresses; } @@ -360,7 +359,7 @@ class NodeDiscoveryBroadcaster extends UdpChannelReplicator { * @throws java.io.IOException */ NodeDiscoveryBroadcaster( - final UdpConfig replicationConfig, + final UdpReplicationConfig replicationConfig, final int serializedEntrySize, final BytesMarshallable externalizable) throws IOException { diff --git a/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java index 2f906a84c..0c7880724 100644 --- a/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java @@ -25,8 +25,6 @@ public final class OffHeapUpdatableChronicleMapBuilder public static OffHeapUpdatableChronicleMapBuilder of( @NotNull Class keyClass, @NotNull Class valueClass) { - if (keyClass != CharSequence.class && keyClass.isInterface()) - keyClass = DataValueClasses.directClassFor(keyClass); if (!offHeapReference(valueClass)) { if (!valueClass.isInterface()) { throw new IllegalArgumentException( diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 1041f7322..b4664a1c5 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -26,7 +26,7 @@ import java.io.Closeable; /** * @author Rob Austin. */ -interface Replica extends Closeable { +public interface Replica extends Closeable { /** * Provides the unique Identifier associated with this map instance.

An identifier is used to diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index c865ed036..88abf77df 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -19,7 +19,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TimeProvider; +import net.openhft.chronicle.hash.TimeProvider; import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.chronicle.hash.serialization.MetaBytesInterop; import net.openhft.chronicle.hash.serialization.MetaBytesWriter; @@ -82,7 +82,7 @@ import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND; * @param the entries key type * @param the entries value type */ -final class ReplicatedChronicleMap, +class ReplicatedChronicleMap, V, VW, MVW extends MetaBytesWriter> extends VanillaChronicleMap implements ChronicleMap, @@ -102,13 +102,12 @@ final class ReplicatedChronicleMap, private transient ModificationDelegator modificationDelegator; private transient long startOfModificationIterators; - public ReplicatedChronicleMap(@NotNull AbstractChronicleMapBuilder builder, - byte identifier) + public ReplicatedChronicleMap(@NotNull AbstractChronicleMapBuilder builder) throws IOException { super(builder); this.timeProvider = builder.timeProvider(); - this.localIdentifier = identifier; + this.localIdentifier = builder.identifier(); if (localIdentifier == -1) { throw new IllegalStateException("localIdentifier should not be -1"); @@ -633,6 +632,8 @@ final class ReplicatedChronicleMap, V acquireWithoutLock(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue, long hash2, boolean create, long timestamp) { + + long keySize = metaKeyInterop.size(keyInterop, key); MultiStoreBytes entry = acquireTmpBytes(); long offset = searchKey(keyInterop, metaKeyInterop, key, keySize, hash2, entry, diff --git a/src/main/java/net/openhft/chronicle/map/Replicator.java b/src/main/java/net/openhft/chronicle/map/Replicator.java index 7580d519d..7d84ae262 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicator.java +++ b/src/main/java/net/openhft/chronicle/map/Replicator.java @@ -19,7 +19,7 @@ package net.openhft.chronicle.map; import java.io.Closeable; import java.io.IOException; -abstract class Replicator { +public abstract class Replicator { /** * A constructor for use in subclasses. diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index b62683c44..6b97d0ee6 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -16,8 +16,8 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.TcpReplicationConfig; +import net.openhft.chronicle.hash.UdpReplicationConfig; import org.jetbrains.annotations.NotNull; import java.io.Closeable; @@ -25,18 +25,10 @@ import java.io.IOException; final class Replicators { - static final String ONLY_UDP_WARN_MESSAGE = - "MISSING TCP REPLICATION : The UdpReplicator only attempts to read data " + - "(it does not enforce or guarantee delivery), you should use" + - "the UdpReplicator if you have a large number of nodes, and you wish" + - "to receive the data before it becomes available on TCP/IP. Since data" + - "delivery is not guaranteed, it is recommended that you only use" + - "the UDP Replicator in conjunction with a TCP Replicator"; - private Replicators() { } - static Replicator tcp(final TcpConfig replicationConfig) { + static Replicator tcp(final TcpReplicationConfig replicationConfig) { return new Replicator() { @Override @@ -50,17 +42,16 @@ final class Replicators { .keyBuilder, builder.valueBuilder); StatelessServerConnector statelessServer = new StatelessServerConnector - (keyValueSerializer, (VanillaChronicleMap) chronicleMap, - builder.entrySize(true)); + (keyValueSerializer, (VanillaChronicleMap) chronicleMap, builder.entrySize()); return new TcpReplicator(replica, entryExternalizable, replicationConfig, - builder.entrySize(true), statelessServer); + builder.entrySize(), statelessServer); } }; } static Replicator udp( - final UdpConfig replicationConfig) { + final UdpReplicationConfig replicationConfig) { return new Replicator() { @Override @@ -70,7 +61,7 @@ final class Replicators { final ChronicleMap chronicleMap) throws IOException { return new UdpReplicator(map, entryExternalizable, replicationConfig, - builder.entrySize(true)); + builder.entrySize()); } }; } diff --git a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java index f69682bcf..9d8b0a740 100644 --- a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java @@ -83,29 +83,18 @@ final class SerializationBuilder implements Cloneable { NullObjectFactory.of(); if (concreteClass(eClass) && Byteable.class.isAssignableFrom(eClass)) { - ByteableMarshaller agileBytesMarshaller = ByteableMarshaller.of((Class) eClass); - sizeMarshaller(agileBytesMarshaller) - .reader(agileBytesMarshaller, factory) - .interop(agileBytesMarshaller); + agileMarshaller(ByteableMarshaller.of((Class) eClass), factory); } else if (eClass == CharSequence.class || eClass == String.class) { reader((BytesReader) CharSequenceReader.of(), factory); writer((BytesWriter) CharSequenceWriter.instance()); } else if (eClass == Void.class) { - sizeMarshaller(VoidMarshaller.INSTANCE) - .reader((BytesReader) VoidMarshaller.INSTANCE, factory) - .interop((BytesInterop) VoidMarshaller.INSTANCE); + agileMarshaller((AgileBytesMarshaller) VoidMarshaller.INSTANCE, factory); } else if (eClass == Long.class) { - sizeMarshaller(LongMarshaller.INSTANCE) - .reader((BytesReader) LongMarshaller.INSTANCE, factory) - .interop((BytesInterop) LongMarshaller.INSTANCE); + agileMarshaller((AgileBytesMarshaller) LongMarshaller.INSTANCE, factory); } else if (eClass == Double.class) { - sizeMarshaller(DoubleMarshaller.INSTANCE) - .reader((BytesReader) DoubleMarshaller.INSTANCE, factory) - .interop((BytesInterop) DoubleMarshaller.INSTANCE); + agileMarshaller((AgileBytesMarshaller) DoubleMarshaller.INSTANCE, factory); } else if (eClass == Integer.class) { - sizeMarshaller(IntegerMarshaller.INSTANCE) - .reader((BytesReader) IntegerMarshaller.INSTANCE, factory) - .interop((BytesInterop) IntegerMarshaller.INSTANCE); + agileMarshaller((AgileBytesMarshaller) IntegerMarshaller.INSTANCE, factory); } else if (eClass == byte[].class) { reader((BytesReader) ByteArrayMarshaller.INSTANCE, factory); interop((BytesInterop) ByteArrayMarshaller.INSTANCE); @@ -136,6 +125,14 @@ final class SerializationBuilder implements Cloneable { return this; } + public SerializationBuilder agileMarshaller(AgileBytesMarshaller agileBytesMarshaller, + ObjectFactory factory) { + if (factory == null) factory = this.factory; + return sizeMarshaller(agileBytesMarshaller) + .reader(agileBytesMarshaller, factory) + .interop(agileBytesMarshaller); + } + public SerializationBuilder interop(BytesInterop interop) { return copyingInterop(null) .setInterop(interop) @@ -145,8 +142,6 @@ final class SerializationBuilder implements Cloneable { } public SerializationBuilder writer(BytesWriter writer) { - if (writer instanceof BytesInterop) - return interop((BytesInterop) writer); return copyingInterop(CopyingInterop.FROM_WRITER) .setInterop(writer) .metaInterop(CopyingMetaBytesInterop @@ -292,19 +287,10 @@ final class SerializationBuilder implements Cloneable { " value don't use object factory"); } else if (interop instanceof ByteableMarshaller) { if (factory instanceof AllocateInstanceObjectFactory) { - ByteableMarshaller agileBytesMarshaller = - ByteableMarshaller.of((Class) eClass); - sizeMarshaller(agileBytesMarshaller) - .reader(agileBytesMarshaller, factory) - .interop(agileBytesMarshaller); + agileMarshaller(ByteableMarshaller.of((Class) eClass), factory); } else { - ByteableMarshaller agileBytesMarshaller = - ByteableMarshaller.of((Class) eClass, (ObjectFactory) factory); - ObjectFactory factory1 = factory; - if (factory1 == null) factory1 = this.factory; - sizeMarshaller(agileBytesMarshaller) - .reader(agileBytesMarshaller, factory1) - .interop(agileBytesMarshaller); + agileMarshaller(ByteableMarshaller.of((Class) eClass, (ObjectFactory) factory), + factory); } } else if (interop instanceof BytesMarshallableMarshaller) { if (factory instanceof AllocateInstanceObjectFactory) { diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index c74f1bc1a..531e118ac 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -19,6 +19,7 @@ package net.openhft.chronicle.map; import com.sun.jdi.connect.spi.ClosedConnectionException; +import net.openhft.chronicle.hash.StatelessBuilder; import net.openhft.chronicle.hash.exceptions.IORuntimeException; import net.openhft.chronicle.hash.exceptions.TimeoutRuntimeException; import net.openhft.lang.io.ByteBufferBytes; @@ -60,13 +61,14 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private volatile SocketChannel clientChannel; private CloseablesManager closeables; - private final StatelessMapConfig builder; + private final StatelessBuilder builder; private int maxEntrySize; private final Class kClass; private final Class vClass; private final boolean putReturnsNull; private final boolean removeReturnsNull; + static enum EventId { HEARTBEAT, STATEFUL_UPDATE, @@ -90,20 +92,20 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { TO_STRING, PUT_ALL, PUT_ALL_WITHOUT_ACC, - HASH_CODE, - ACQUIRE_USING + HASH_CODE } private long transactionID; - StatelessChronicleMap(final StatelessMapConfig builder, + StatelessChronicleMap(final StatelessBuilder builder, final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { + this.builder = builder; this.keyValueSerializer = new KeyValueSerializer(chronicleMapBuilder.keyBuilder, chronicleMapBuilder.valueBuilder); this.putReturnsNull = chronicleMapBuilder.putReturnsNull(); this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull(); - this.maxEntrySize = chronicleMapBuilder.entrySize(false); + this.maxEntrySize = chronicleMapBuilder.entrySize(); this.vClass = chronicleMapBuilder.valueBuilder.eClass; this.kClass = chronicleMapBuilder.keyBuilder.eClass; // this.putReturnsNull = putReturnsNull; @@ -393,11 +395,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } - /** - * @param key the key whose associated value is to be returned - * @param usingValue the object to read value data in, if possible - * @return - */ public synchronized V acquireUsing(K key, V usingValue) { throw new UnsupportedOperationException("acquireUsing() is not supported for stateless " + "clients"); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index d08f361d3..087e44180 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -18,9 +18,8 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.IdentifierListener; -import net.openhft.chronicle.hash.replication.TcpConfig; -import net.openhft.chronicle.hash.replication.ThrottlingConfig; +import net.openhft.chronicle.hash.TcpReplicationConfig; +import net.openhft.chronicle.hash.ThrottlingConfig; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; import net.openhft.lang.threadlocal.ThreadLocalCopies; @@ -52,12 +51,12 @@ interface Work { } /** - * Used with a {@link net.openhft.chronicle.map.ReplicatedChronicleMap} to send data between the - * maps using a socket connection {@link net.openhft.chronicle.map.TcpReplicator} + * Used with a {@link net.openhft.chronicle.map.ReplicatedChronicleMap} to send data between the maps using a socket + * connection {@link net.openhft.chronicle.map.TcpReplicator} * * @author Rob Austin. */ -final class TcpReplicator extends AbstractChannelReplicator implements Closeable { +public class TcpReplicator extends AbstractChannelReplicator implements Closeable { public static final int STATELESS_CLIENT = -127; public static final byte NOT_SET = (byte) HEARTBEAT.ordinal(); @@ -74,7 +73,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private final byte localIdentifier; private final int maxEntrySizeBytes; private final Replica.EntryExternalizable externalizable; - private final TcpConfig replicationConfig; + private final TcpReplicationConfig replicationConfig; private final StatelessServerConnector statelessServerConnector; private long selectorTimeout; @@ -86,11 +85,11 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * the entry, but setting it too large reduces the workable * space in the buffer. * @param statelessServerConnector set to NULL if not required - * @throws IOException on an io error. + * @throws IOException on an io error. */ public TcpReplicator(@NotNull final Replica replica, @NotNull final Replica.EntryExternalizable externalizable, - @NotNull final TcpConfig replicationConfig, + @NotNull final TcpReplicationConfig replicationConfig, final int maxEntrySizeBytes, @Nullable StatelessServerConnector statelessServerConnector) throws IOException { @@ -279,8 +278,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * check to see if we have lost connection with the remote node and if we have attempts a - * reconnect. + * check to see if we have lost connection with the remote node and if we have attempts a reconnect. * * @param key the key relating to the heartbeat that we are checking * @param approxTimeOutTime the approximate time in milliseconds @@ -427,8 +425,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear - * all the old bootstrap information + * this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear all the old + * bootstrap information * * @param key the nio SelectionKey */ @@ -440,8 +438,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * used to exchange identifiers and timestamps and heartbeat intervals between the server and - * client + * used to exchange identifiers and timestamps and heartbeat intervals between the server and client * * @param key the SelectionKey relating to the this cha * @param socketChannel @@ -481,8 +478,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // this can occur sometimes when if 2 or more remote node attempt to use node discovery at the // same time - final IdentifierListener identifierListener = - replicationConfig.nonUniqueIdentifierListener(); + final IdentifierListener identifierListener = replicationConfig.identifierListener; final SocketAddress remoteAddress = socketChannel.getRemoteAddress(); @@ -622,10 +618,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * sets interestOps to "selector keys",The change to interestOps much be on the same thread as - * the selector. This class, allows via {@link AbstractChannelReplicator - * .KeyInterestUpdater#set(int)} to holds a pending change in interestOps ( via a bitset ), - * this change is processed later on the same thread as the selector + * sets interestOps to "selector keys",The change to interestOps much be on the same thread as the selector. This + * class, allows via {@link AbstractChannelReplicator .KeyInterestUpdater#set(int)} to holds a pending change in + * interestOps ( via a bitset ), this change is processed later on the same thread as the selector */ private static class KeyInterestUpdater { @@ -656,8 +651,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * @param keyIndex the index of the key that has changed, the list of keys is provided by - * the constructor {@link KeyInterestUpdater(int, SelectionKey[])} + * @param keyIndex the index of the key that has changed, the list of keys is provided by the constructor {@link + * KeyInterestUpdater(int, SelectionKey[])} */ public void set(int keyIndex) { changeOfOpWriteRequired.set(keyIndex); @@ -898,8 +893,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * writes all the entries that have changed, to the buffer which will later be written to - * TCP/IP + * writes all the entries that have changed, to the buffer which will later be written to TCP/IP * * @param modificationIterator a record of which entries have modification * @param selectionKey @@ -977,8 +971,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * used to send an single zero byte if we have not send any data for up to the - * localHeartbeatInterval + * used to send an single zero byte if we have not send any data for up to the localHeartbeatInterval */ private void writeHeartbeatToBuffer() { @@ -995,8 +988,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable /** - * removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE - * will get added back in as soon as we have data to write + * removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE will get added back + * in as soon as we have data to write * * @param socketChannel the socketChannel we wish to stop writing to * @param attached data associated with the socketChannels key @@ -1284,7 +1277,6 @@ class StatelessServerConnector { switch (event) { - case LONG_SIZE: return longSize(reader, writer, sizeLocation); @@ -1300,10 +1292,6 @@ class StatelessServerConnector { case GET: return get(reader, writer, sizeLocation); - case ACQUIRE_USING: - return acquireUsing(reader, writer, sizeLocation); - - case PUT: return put(reader, writer, sizeLocation); @@ -1488,26 +1476,6 @@ class StatelessServerConnector { return null; } - - private Work acquireUsing(Bytes reader, Bytes writer, final long sizeLocation) { - if (MAP_SUPPORTS_BYTES) { - map.acquireUsing(reader, writer); - } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K k = readKey(reader, local); - final V v = readValue(reader, local); - - try { - writeValue(map.acquireUsing(k, v), writer, local); - } catch (RuntimeException e) { - return sendException(writer, sizeLocation, e); - } - } - writeSizeAndFlags(sizeLocation, false, writer); - return null; - - } - private Work get(Bytes reader, Bytes writer, final long sizeLocation) { if (MAP_SUPPORTS_BYTES) { map.get(reader, writer); diff --git a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java index 2896832d2..337e48336 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.UdpReplicationConfig; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -69,7 +69,7 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. * @param localIdentifier just used for logging * @throws IOException */ - UdpChannelReplicator(@NotNull final UdpConfig replicationConfig, + UdpChannelReplicator(@NotNull final UdpReplicationConfig replicationConfig, final int maxPayloadSize, final byte localIdentifier) throws IOException { diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index 504804982..ed43075dd 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.UdpReplicationConfig; import net.openhft.lang.io.ByteBufferBytes; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; @@ -36,14 +36,14 @@ import java.nio.channels.DatagramChannel; * it becomes available on TCP/IP. In order to not miss data, UdpReplicator should be used in conjunction with * the TCP Replicator. */ -final class UdpReplicator extends UdpChannelReplicator implements Replica.ModificationNotifier, Closeable { +public class UdpReplicator extends UdpChannelReplicator implements Replica.ModificationNotifier, Closeable { private static final Logger LOG = LoggerFactory.getLogger(UdpReplicator.class.getName()); public UdpReplicator(@NotNull final Replica replica, @NotNull final Replica.EntryExternalizable entryExternalizable, - @NotNull final UdpConfig replicationConfig, + @NotNull final UdpReplicationConfig replicationConfig, final int serializedEntrySize) throws IOException { diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 8bfbea60e..f827a5e8f 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -41,6 +41,7 @@ import java.nio.channels.FileChannel; import java.util.*; import java.util.concurrent.TimeUnit; +import static java.lang.Long.numberOfTrailingZeros; import static java.lang.Math.max; import static java.lang.Thread.currentThread; import static net.openhft.lang.MemoryUnit.*; @@ -128,8 +129,8 @@ class VanillaChronicleMap, lockTimeOutNS = builder.lockTimeOut(TimeUnit.NANOSECONDS); - boolean replicated = getClass() == ReplicatedChronicleMap.class; - this.entrySize = builder.entrySize(replicated); + // this.replicas = builder.replicas(); + this.entrySize = builder.entrySize(); this.alignment = builder.entryAndValueAlignment(); this.errorListener = builder.errorListener(); @@ -275,7 +276,19 @@ class VanillaChronicleMap, + sizeOfEntriesInSegment(); if ((ss & 63L) != 0) throw new AssertionError(); - return ss; + return breakL1CacheAssociativityContention(ss); + } + + private long breakL1CacheAssociativityContention(long segmentSize) { + // Conventional alignment to break is 4096 (given Intel's 32KB 8-way L1 cache), + // for any case break 2 times smaller alignment + int alignmentToBreak = 2048; + int eachNthSegmentFallIntoTheSameSet = + max(1, alignmentToBreak >> numberOfTrailingZeros(segmentSize)); + if (eachNthSegmentFallIntoTheSameSet < actualSegments) { + segmentSize |= 64L; // make segment size "odd" (in cache lines) + } + return segmentSize; } int multiMapsPerSegment() { @@ -584,10 +597,6 @@ class VanillaChronicleMap, throw new UnsupportedOperationException("todo"); } - public V acquireUsing(Bytes reader, Bytes writer) { - throw new UnsupportedOperationException("todo"); - } - void get(Bytes reader, Bytes writer) { throw new UnsupportedOperationException("todo"); } @@ -1406,7 +1415,8 @@ class VanillaChronicleMap, @Override public void removeEntry() { - removeWithoutLock(copies, metaKeyInterop, keyInterop, key(), null, segmentHash); + if (copies != null) + removeWithoutLock(copies, metaKeyInterop, keyInterop, key(), null, segmentHash); } } @@ -1420,7 +1430,9 @@ class VanillaChronicleMap, putWithoutLock(copies, metaKeyInterop, keyInterop, key(), value(), segmentHash, true); + copies = null; putOnClose = true; + Segment.this.writeUnlock(); } diff --git a/src/main/java/net/openhft/chronicle/map/WriteContext.java b/src/main/java/net/openhft/chronicle/map/WriteContext.java index ba5471903..26cc84baf 100644 --- a/src/main/java/net/openhft/chronicle/map/WriteContext.java +++ b/src/main/java/net/openhft/chronicle/map/WriteContext.java @@ -21,9 +21,10 @@ package net.openhft.chronicle.map; * @author Rob Austin. */ public interface WriteContext extends Context { - void dontPutOnClose(); void removeEntry(); + // todo I have no idea why we have this, as this will always be done on when calling close() + // void update(); } diff --git a/src/main/java/net/openhft/chronicle/map/ZeroOutValueBytes.java b/src/main/java/net/openhft/chronicle/map/ZeroOutValueBytes.java index 668e976a6..a6e2ce845 100644 --- a/src/main/java/net/openhft/chronicle/map/ZeroOutValueBytes.java +++ b/src/main/java/net/openhft/chronicle/map/ZeroOutValueBytes.java @@ -29,7 +29,7 @@ final class ZeroOutValueBytes implements PrepareValueBytes { @Override public void prepare(Bytes bytes, K key) { long pos = bytes.position(); - bytes.zeroOut(pos, pos + valueSize, true); + bytes.zeroOut(pos, pos + valueSize); bytes.skip(valueSize); } } diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index fd607424c..e4c6bf44e 100644 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -19,9 +19,7 @@ package net.openhft.chronicle.set; import net.openhft.chronicle.hash.*; -import net.openhft.chronicle.hash.replication.SimpleReplication; -import net.openhft.chronicle.hash.replication.TcpConfig; -import net.openhft.chronicle.hash.replication.TimeProvider; +import net.openhft.chronicle.map.Alignment; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import net.openhft.lang.io.serialization.BytesMarshaller; @@ -32,7 +30,6 @@ import org.jetbrains.annotations.NotNull; import java.io.File; import java.io.IOException; -import java.net.InetSocketAddress; import java.util.concurrent.TimeUnit; /** @@ -53,7 +50,7 @@ public class ChronicleSetBuilder ChronicleSetBuilder(Class keyClass) { chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, DummyValue.class) - .valueMarshallers(DummyValueMarshaller.INSTANCE, DummyValueMarshaller.INSTANCE); + .valueMarshaller(DummyValueMarshaller.INSTANCE); } public static ChronicleSetBuilder of(Class keyClass) { @@ -170,6 +167,25 @@ public class ChronicleSetBuilder return this; } + @Override + public ChronicleSetBuilder replicators(byte identifier, ReplicationConfig... replicators) { + chronicleMapBuilder.replicators(identifier, replicators); + return this; + } + + @Override + public ChronicleSetBuilder channel(ChannelProvider.ChronicleChannel chronicleChannel) { + chronicleMapBuilder.channel(chronicleChannel); + return this; + } + + @Override + public ChronicleSetBuilder disableReplication() { + chronicleMapBuilder.disableReplication(); + return this; + } + + @Override public String toString() { return " ChronicleSetBuilder{" + @@ -249,39 +265,23 @@ public class ChronicleSetBuilder } @Override - public ChronicleSetBuilder replication(SimpleReplication replication) { - chronicleMapBuilder.replication(replication); + public ChronicleSetBuilder stateless(StatelessBuilder statelessBuilder) { + chronicleMapBuilder.stateless(statelessBuilder); return this; } @Override - public ChronicleSetBuilder replication(byte identifier, TcpConfig tcpTransportAndNetwork) { - chronicleMapBuilder.replication(identifier, tcpTransportAndNetwork); + public ChronicleSetBuilder file(File file) throws IOException { + chronicleMapBuilder.file(file); return this; } - @Override - public StatelessClientConfig> statelessClient( - InetSocketAddress remoteAddress) { - return new StatelessSetConfig<>(chronicleMapBuilder.statelessClient(remoteAddress)); - } - - @Override - public ChronicleHashInstanceConfig> instance() { - return new InstanceConfig<>(chronicleMapBuilder.instance()); - } - @Override public ChronicleSet create() throws IOException { final ChronicleMap map = chronicleMapBuilder.create(); return new SetFromMap(map); } - @Override - public ChronicleSet createPersistedTo(File file) throws IOException { - ChronicleMap map = chronicleMapBuilder.createPersistedTo(file); - return new SetFromMap(map); - } } diff --git a/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java b/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java index ddb3956f7..3ec1e79bb 100644 --- a/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java +++ b/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java @@ -18,15 +18,12 @@ package net.openhft.chronicle.set; -import net.openhft.chronicle.hash.serialization.BytesInterop; -import net.openhft.chronicle.hash.serialization.BytesReader; -import net.openhft.chronicle.hash.serialization.SizeMarshaller; +import net.openhft.chronicle.hash.serialization.AgileBytesMarshaller; import net.openhft.lang.io.Bytes; import static net.openhft.chronicle.set.DummyValue.DUMMY_VALUE; -enum DummyValueMarshaller - implements BytesInterop, BytesReader, SizeMarshaller { +enum DummyValueMarshaller implements AgileBytesMarshaller { INSTANCE; @Override @@ -45,7 +42,7 @@ enum DummyValueMarshaller } @Override - public DummyValue read(Bytes bytes, long size, DummyValue toReuse) { + public DummyValue read(Bytes bytes, long size, DummyValue dummyValue) { return DUMMY_VALUE; } diff --git a/src/test/java/eg/BigData.java b/src/test/java/eg/BigData.java index 177b1059f..745998a78 100644 --- a/src/test/java/eg/BigData.java +++ b/src/test/java/eg/BigData.java @@ -52,7 +52,7 @@ public class BigData { String chmPath = dir + "/testmap-" + Long.toString(System.nanoTime(), 36); new File(chmPath).deleteOnExit(); try { - theMap = builder.createPersistedTo(new File(chmPath)); + theMap = builder.file(new File(chmPath)).create(); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); diff --git a/src/test/java/net/openhft/chronicle/map/AcquireGetUsingMain.java b/src/test/java/net/openhft/chronicle/map/AcquireGetUsingMain.java index fe6b10497..befa4148a 100644 --- a/src/test/java/net/openhft/chronicle/map/AcquireGetUsingMain.java +++ b/src/test/java/net/openhft/chronicle/map/AcquireGetUsingMain.java @@ -29,9 +29,7 @@ import static org.junit.Assert.fail; public class AcquireGetUsingMain { public static void main(String[] args) throws Exception { File file = new File(System.getProperty("java.io.tmpdir") + "/test1"); - ChronicleMap theSharedMap = - OffHeapUpdatableChronicleMapBuilder.of(String.class, Data.class) - .createPersistedTo(file); + ChronicleMap theSharedMap = OffHeapUpdatableChronicleMapBuilder.of(String.class, Data.class).file(file).create(); Data data = DataValueClasses.newDirectReference(Data.class); String processType = "testkey"; if (theSharedMap.getUsing(processType, data) == null) { diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java index 41db8cced..1225a0fc9 100644 --- a/src/test/java/net/openhft/chronicle/map/Builder.java +++ b/src/test/java/net/openhft/chronicle/map/Builder.java @@ -19,13 +19,12 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.TcpReplicationConfig; import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; import java.net.InetSocketAddress; -import java.util.Arrays; import java.util.concurrent.TimeUnit; /** @@ -72,10 +71,10 @@ public class Builder { final byte identifier, final int serverPort, final InetSocketAddress... endpoints) throws IOException { - TcpConfig tcpConfig = TcpConfig.unknownTopology(serverPort, Arrays.asList(endpoints)) + TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(serverPort, endpoints) .heartBeatInterval(1L, TimeUnit.SECONDS); return ChronicleMapBuilder.of(kClass, vClass) - .replication(identifier, tcpConfig); + .replicators(identifier, tcpConfig); } diff --git a/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java b/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java index 6649c1d86..e6711b4ca 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java +++ b/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java @@ -79,7 +79,7 @@ public class CHMLatencyTestMain { OffHeapUpdatableChronicleMapBuilder.of(LongValue.class, LongValue.class) .entries(KEYS) .entrySize(16) - .createPersistedTo(file); + .createWithFile(file); // add keys LongValue key = DataValueClasses.newInstance(LongValue.class); diff --git a/src/test/java/net/openhft/chronicle/map/CHMTest5.java b/src/test/java/net/openhft/chronicle/map/CHMTest5.java index 6979bc473..e631c45be 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMTest5.java +++ b/src/test/java/net/openhft/chronicle/map/CHMTest5.java @@ -44,8 +44,7 @@ public class CHMTest5 { String chmPath = System.getProperty("java.io.tmpdir") + System.getProperty("file.separator") + "CHMTest5"; - ChronicleMap theSharedMap = - builder.createPersistedTo(new File(chmPath)); + ChronicleMap theSharedMap = builder.file(new File(chmPath)).create(); //Now get the shared data object, auto-creating if it's not there CHMTest5Data data = DataValueClasses.newDirectReference(CHMTest5Data.class); diff --git a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java index ed0b5cb2f..c726da6eb 100644 --- a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java @@ -19,8 +19,9 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.ChannelProvider; +import net.openhft.chronicle.hash.ChannelProviderBuilder; +import net.openhft.chronicle.hash.TcpReplicationConfig; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -46,44 +47,44 @@ public class ChannelReplicationTest { private ChronicleMap map1b; private ChronicleMap map2b; - private ReplicationHub hubA; - private ReplicationHub hubB; + private ChannelProvider channelProviderA; + private ChannelProvider channelProviderB; @Before public void setup() throws IOException { { - TcpConfig tcpConfig = TcpConfig - .forSendingNode(8086, new InetSocketAddress("localhost", 8087)) + TcpReplicationConfig tcpConfig = TcpReplicationConfig + .of(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); - hubA = ReplicationHub.builder() - .tcpTransportAndNetwork(tcpConfig) - .create((byte) 1); + channelProviderA = new ChannelProviderBuilder() + .replicators((byte) 1, tcpConfig) + .create(); map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); + .channel(channelProviderA.createChannel((short) 1)).create(); } { - TcpConfig tcpConfig = - TcpConfig.forReceivingOnlyNode(8087).heartBeatInterval(1, SECONDS); + TcpReplicationConfig tcpConfig = + TcpReplicationConfig.of(8087).heartBeatInterval(1, SECONDS); - hubB = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig) - .create((byte) 2); + channelProviderB = new ChannelProviderBuilder().replicators((byte) 2, tcpConfig) + .create(); map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .instance().replicatedViaChannel(hubB.createChannel((short) 1)).create(); + .channel(channelProviderB.createChannel((short) 1)).create(); } } @After public void tearDown() throws InterruptedException { - for (final Closeable closeable : new Closeable[]{map1a, map1b, map2a, map2b}) { + for (final Closeable closeable : new Closeable[]{channelProviderA, channelProviderB}) { try { closeable.close(); } catch (IOException e) { @@ -98,12 +99,12 @@ public class ChannelReplicationTest { map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .instance().replicatedViaChannel(hubB.createChannel((short) 2)).create(); + .channel(channelProviderB.createChannel((short) 2)).create(); map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .instance().replicatedViaChannel(hubA.createChannel((short) 2)).create(); + .channel(channelProviderA.createChannel((short) 2)).create(); map2a.put(1, "EXAMPLE-2"); map1a.put(1, "EXAMPLE-1"); diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index 294304c45..091b097f8 100644 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -529,79 +529,6 @@ public class ChronicleMapTest { map.close(); } - @Test - @Ignore - public void testLeanEntrySizeInt() throws IOException { - File file = File.createTempFile("testLeanEntrySize-Int", ".deleteme"); - file.deleteOnExit(); - int entries = 1000000; - int entrySize = 8; - ChronicleMap map = OffHeapUpdatableChronicleMapBuilder - .of(IntValue.class, IntValue.class) - .entrySize(entrySize) - .entries(entries) - .create(); - IntValue key = DataValueClasses.newInstance(IntValue.class); - IntValue value = DataValueClasses.newDirectReference(IntValue.class); - for (int i = 0; i < entries; i++) { - key.setValue(i); - try (WriteContext wc = map.acquireUsingLocked(key, value)) { - value.setValue(i); - } - } - for (int i = 0; i < entries; i++) { - key.setValue(i); - try (ReadContext rc = map.getUsingLocked(key, value)) { - assertTrue(rc.present()); - assertEquals(i, value.getValue()); - } - } - map.close(); - - // check the extents of the map. - int hashSize = 8; // assume short-short hashing - int sizePerEntry = hashSize * 2 + entrySize * 5 / 4 /* + 25% */; - long fileSizeExpected = entries * sizePerEntry; - long length = file.length(); - assertTrue(length > fileSizeExpected); - assertTrue(length < fileSizeExpected * 12 / 10); - } - - @Test - @Ignore - public void testLeanEntrySizeLong() throws IOException { - File file = File.createTempFile("testLeanEntrySize-Long", ".deleteme"); - file.deleteOnExit(); - int entries = 1000000; - int entrySize = 16; - ChronicleMap map = OffHeapUpdatableChronicleMapBuilder - .of(LongValue.class, LongValue.class) - .entrySize(entrySize) - .entries(entries) - .create(); - LongValue key = DataValueClasses.newInstance(LongValue.class); - LongValue value = DataValueClasses.newDirectReference(LongValue.class); - for (long i = 0; i < entries; i++) { - key.setValue(i); - map.acquireUsing(key, value); - value.setValue(i); - } - for (long i = 0; i < entries; i++) { - key.setValue(i); - assertNotNull(map.getUsing(key, value)); - assertEquals(i, value.getValue()); - } - map.close(); - - // check the extents of the map. - int hashSize = 8; // assume short-short hashing - int sizePerEntry = hashSize * 2 + entrySize * 5 / 4 /* + 25% */; - long fileSizeExpected = entries * sizePerEntry; - long length = file.length(); - assertTrue(length > fileSizeExpected); - assertTrue(length < fileSizeExpected * 12 / 10); - } - @Test public void testAcquireFromMultipleThreads() throws Exception { int entries = 1000 * 1000; @@ -651,19 +578,19 @@ public class ChronicleMapTest { int procs = Runtime.getRuntime().availableProcessors(); int threads = procs * 2; // runs > 100 ? procs / 2 : procs; ExecutorService es = Executors.newFixedThreadPool(procs); - for (int runs : new int[]{10, 25, 50, 250, 500, 1000, 2500}) { + for (int runs : new int[]{10, 50, 250, 500, 1000, 2500}) { // JAVA 8 produces more garbage than previous versions for internal work. // System.gc(); final long entries = runs * 1000 * 1000L; - OffHeapUpdatableChronicleMapBuilder builder = OffHeapUpdatableChronicleMapBuilder - .of(LongValue.class, LongValue.class) + OffHeapUpdatableChronicleMapBuilder builder = OffHeapUpdatableChronicleMapBuilder + .of(CharSequence.class, LongValue.class) .entries(entries) .entryAndValueAlignment(OF_8_BYTES) - .entrySize(16); + .entrySize(24); File tmpFile = File.createTempFile("testAcquirePerf", ".deleteme"); tmpFile.deleteOnExit(); - final ChronicleMap map = builder.createPersistedTo(tmpFile); + final ChronicleMap map = builder.createWithFile(tmpFile); int count = runs > 500 ? runs > 1200 ? 3 : 5 : 5; final int independence = Math.min(procs, runs > 500 ? 8 : 4); @@ -676,14 +603,16 @@ public class ChronicleMapTest { futures.add(es.submit(new Runnable() { @Override public void run() { - LongValue key = DataValueClasses.newDirectInstance(LongValue.class); LongValue value = nativeLongValue(); + StringBuilder sb = new StringBuilder(); long next = 50 * 1000 * 1000; // use a factor to give up to 10 digit numbers. int factor = Math.max(1, (int) ((10 * 1000 * 1000 * 1000L - 1) / entries)); for (long j = t % independence; j < entries + independence - 1; j += independence) { - key.setValue(j * factor); - map.acquireUsing(key, value); + sb.setLength(0); + sb.append("us:"); + sb.append(j * factor); + map.acquireUsing(sb, value); long n = value.addAtomicValue(1); assert n > 0 && n < 1000 : "Counter corrupted " + n; if (t == 0 && j >= next) { @@ -710,7 +639,6 @@ public class ChronicleMapTest { es.shutdown(); es.awaitTermination(1, TimeUnit.MINUTES); } - @Test @Ignore public void testAcquireLockedPerf() @@ -732,7 +660,7 @@ public class ChronicleMapTest { File tmpFile = File.createTempFile("testAcquirePerf", ".deleteme"); tmpFile.deleteOnExit(); - final ChronicleMap map = builder.createPersistedTo(tmpFile); + final ChronicleMap map = builder.createWithFile(tmpFile); int count = runs > 500 ? runs > 1200 ? 3 : 5 : 5; final int independence = Math.min(procs, runs > 500 ? 8 : 4); @@ -755,7 +683,7 @@ public class ChronicleMapTest { sb.append("us:"); sb.append(j * factor); long n; - try (WriteContext wc = map.acquireUsingLocked(sb, value)) { + try(WriteContext wc = map.acquireUsingLocked(sb, value)) { n = value.addValue(1); } assert n > 0 && n < 1000 : "Counter corrupted " + n; @@ -1301,9 +1229,10 @@ public class ChronicleMapTest { .entries(10) .minSegments(1) .entrySize(10); + builder.file(getPersistenceFile()); testOversizeEntriesPutRemoveReplace( (VanillaChronicleMap) - builder.createPersistedTo(getPersistenceFile()) + builder.create() ); } @@ -1405,7 +1334,7 @@ public class ChronicleMapTest { .entrySize(16); File tmpFile = File.createTempFile("testAcquireUsingLocked", ".deleteme"); tmpFile.deleteOnExit(); - final ChronicleMap map = builder.createPersistedTo(tmpFile); + final ChronicleMap map = builder.createWithFile(tmpFile); LongValue value = nativeLongValue(); @@ -1462,7 +1391,7 @@ public class ChronicleMapTest { .entrySize(40); File tmpFile = File.createTempFile("testAcquireUsingLocked", ".deleteme"); tmpFile.deleteOnExit(); - final ChronicleMap map = builder.createPersistedTo(tmpFile); + final ChronicleMap map = builder.createWithFile(tmpFile); LongValue value = DataValueClasses.newInstance(LongValue.class); diff --git a/src/test/java/net/openhft/chronicle/map/DefaultValueTest.java b/src/test/java/net/openhft/chronicle/map/DefaultValueTest.java index 96180ebc8..249b017ad 100644 --- a/src/test/java/net/openhft/chronicle/map/DefaultValueTest.java +++ b/src/test/java/net/openhft/chronicle/map/DefaultValueTest.java @@ -39,7 +39,7 @@ public class DefaultValueTest { defaultValue.add(42); try (ChronicleMap> map = ChronicleMapBuilder .of(String.class, (Class>) ((Class) List.class)) - .defaultValue(defaultValue).createPersistedTo(file)) { + .defaultValue(defaultValue).file(file).create()) { assertEquals(defaultValue, map.acquireUsing("a", null)); assertEquals(1, map.size()); @@ -49,8 +49,8 @@ public class DefaultValueTest { } try (ChronicleMap> map = ChronicleMapBuilder - .of(String.class, (Class>) ((Class) List.class)) - .createPersistedTo(file)) { + .of(String.class, (Class>) ((Class) List.class)).file(file). + create()) { assertEquals(defaultValue, map.acquireUsing("c", null)); } } finally { diff --git a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java index d3cad018a..c15ff7d53 100644 --- a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java @@ -55,8 +55,9 @@ public class EntryCountMapTest { .entries(entries) .minSegments(segments) .entrySize(entrySize) - .entryAndValueAlignment(alignment); - return mapBuilder.createPersistedTo(getPersistenceFile()); + .entryAndValueAlignment(alignment) + .file(getPersistenceFile()); + return mapBuilder.create(); } @Test diff --git a/src/test/java/net/openhft/chronicle/map/OSResizesMain.java b/src/test/java/net/openhft/chronicle/map/OSResizesMain.java index 4298bd5cc..da15a2848 100644 --- a/src/test/java/net/openhft/chronicle/map/OSResizesMain.java +++ b/src/test/java/net/openhft/chronicle/map/OSResizesMain.java @@ -29,17 +29,15 @@ import java.util.Arrays; /** * This example shows that the OS resizes the usage of a CHM as needed. It is not as critical to worry about this. *

- * System memory: 7.7 GB, Extents of map: 100000.0 GB, disk used: 104MB, addressRange: 24e02a7fc000-7fd33c000000 + * System memory: 7.7 GB, Extents of map: 2199.0 GB, disk used: 13MB, addressRange: 7d380b7bd000-7f380c000000 *

*/ public class OSResizesMain { public static void main(String[] args) throws IOException, InterruptedException { File file = File.createTempFile("over-sized", "deleteme"); ChronicleMap map = ChronicleMapBuilder.of(String.class, String.class) - .entrySize(100 * 1000 * 1000) - .entries(1000 * 1000) - .actualSegments(1) - .createPersistedTo(file); + .entrySize(100 * 1024 * 1024) + .entries(1000000).create(); for (int i = 0; i < 10000; i++) { char[] chars = new char[i]; Arrays.fill(chars, '+'); diff --git a/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java b/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java index 2b5f61eff..dcd9890bc 100644 --- a/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java +++ b/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java @@ -19,8 +19,9 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.ChannelProvider; +import net.openhft.chronicle.hash.ChannelProviderBuilder; +import net.openhft.chronicle.hash.TcpReplicationConfig; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -47,8 +48,8 @@ public class PostChannelCreationTest { private ChronicleMap map1b; private ChronicleMap map2b; - private ReplicationHub hubA; - private ReplicationHub hubB; + private ChannelProvider channelProviderA; + private ChannelProvider channelProviderB; public static File getPersistenceFile() { String TMP = System.getProperty("java.io.tmpdir"); @@ -63,40 +64,40 @@ public class PostChannelCreationTest { public void setup() throws IOException { { - final TcpConfig tcpConfig = TcpConfig - .forSendingNode(8086, new InetSocketAddress("localhost", 8087)) + final TcpReplicationConfig tcpConfig = TcpReplicationConfig + .of(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); - hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig) - .create((byte) 1); + channelProviderA = new ChannelProviderBuilder() + .replicators((byte) 1, tcpConfig).create(); // this is how you add maps after the custer is created map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); + .channel(channelProviderA.createChannel((short) 1)).create(); map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .instance().replicatedViaChannel(hubA.createChannel((short) 2)).create(); + .channel(channelProviderA.createChannel((short) 2)).create(); } { - final TcpConfig tcpConfig = TcpConfig - .forSendingNode(8087, new InetSocketAddress("localhost", 8086)) + final TcpReplicationConfig tcpConfig = TcpReplicationConfig + .of(8087, new InetSocketAddress("localhost", 8086)) .heartBeatInterval(1, SECONDS); - hubB = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig) - .create((byte) 2); + channelProviderB = new ChannelProviderBuilder() + .replicators((byte) 2, tcpConfig).create(); // this is how you add maps after the custer is created map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .instance().replicatedViaChannel(hubB.createChannel((short) 1)).create(); + .channel(channelProviderB.createChannel((short) 1)).create(); map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .instance().replicatedViaChannel(hubB.createChannel((short) 2)).create(); + .channel(channelProviderB.createChannel((short) 2)).create(); } @@ -105,7 +106,7 @@ public class PostChannelCreationTest { @After public void tearDown() throws InterruptedException { - for (final Closeable closeable : new Closeable[]{map1a, map1b, map2a, map2b}) { + for (final Closeable closeable : new Closeable[]{channelProviderA, channelProviderB}) { try { closeable.close(); } catch (IOException e) { diff --git a/src/test/java/net/openhft/chronicle/map/SerializerTest.java b/src/test/java/net/openhft/chronicle/map/SerializerTest.java index 4bf4a05e7..5808d89b4 100644 --- a/src/test/java/net/openhft/chronicle/map/SerializerTest.java +++ b/src/test/java/net/openhft/chronicle/map/SerializerTest.java @@ -73,7 +73,7 @@ public class SerializerTest { ByteBufferBytes in = out.slice(); ChronicleMapBuilder builder = ChronicleMapBuilder.of(Integer.class, valueClass); - builder.preMapConstruction(false); + builder.preMapConstruction(); Serializer v = new Serializer(builder.valueBuilder); @@ -96,7 +96,7 @@ public class SerializerTest { ByteBufferBytes in = out.slice(); ChronicleMapBuilder builder = ChronicleMapBuilder.of(clazz, Integer.class); - builder.preMapConstruction(false); + builder.preMapConstruction(); { Serializer v = new Serializer(builder.keyBuilder); diff --git a/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java b/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java index 1d48e9579..8cbd1fe0b 100644 --- a/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java +++ b/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java @@ -19,8 +19,9 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.ChannelProvider; +import net.openhft.chronicle.hash.ChannelProviderBuilder; +import net.openhft.chronicle.hash.TcpReplicationConfig; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -45,8 +46,8 @@ public class SingleMapChannelTest { private ChronicleMap map1b; - private ReplicationHub hubA; - private ReplicationHub hubB; + private ChannelProvider channelProviderA; + private ChannelProvider channelProviderB; public static File getPersistenceFile() { String TMP = System.getProperty("java.io.tmpdir"); @@ -60,41 +61,40 @@ public class SingleMapChannelTest { public void setup() throws IOException { { - final TcpConfig tcpConfig = TcpConfig - .forSendingNode(8086, new InetSocketAddress("localhost", 8087)) + final TcpReplicationConfig tcpReplicationConfig = TcpReplicationConfig + .of(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); int maxEntrySize = 1024; byte identifier= 1; - hubA = ReplicationHub.builder() + channelProviderA = new ChannelProviderBuilder() .maxEntrySize(maxEntrySize) - .tcpTransportAndNetwork(tcpConfig) - .create(identifier); + .replicators(identifier, tcpReplicationConfig).create(); // this is how you add maps after the custer is created map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); + .channel(channelProviderA.createChannel((short) 1)).create(); } { - final TcpConfig tcpConfig = TcpConfig - .forSendingNode(8087, new InetSocketAddress("localhost", 8086)) + final TcpReplicationConfig tcpReplicationConfig = TcpReplicationConfig + .of(8087, new InetSocketAddress("localhost", 8086)) .heartBeatInterval(1, SECONDS); - hubB = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig).create((byte) 2); + channelProviderB = new ChannelProviderBuilder() + .replicators((byte) 2, tcpReplicationConfig).create(); // this is how you add maps after the custer is created map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(1000) - .instance().replicatedViaChannel(hubB.createChannel((short) 1)) - .create(); + .channel(channelProviderB.createChannel((short) 1)) + .entries(1000).create(); } } @After public void tearDown() throws InterruptedException { - for (final Closeable closeable : new Closeable[]{map1a, map1b}) { + for (final Closeable closeable : new Closeable[]{channelProviderA, channelProviderB}) { try { closeable.close(); } catch (IOException e) { diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 35225b056..e3495a017 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.TcpReplicationConfig; import org.junit.Assert; import org.junit.Test; @@ -29,6 +29,8 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; +import static net.openhft.chronicle.hash.StatelessBuilder.remoteAddress; + /** * @author Rob Austin. @@ -39,7 +41,7 @@ public class StatelessClientTest { public static final int SIZE = 100000; - @Test(timeout = 5000) + @Test public void testBufferOverFlowPutAllAndEntrySet() throws IOException, InterruptedException { final ChronicleMap serverMap; @@ -50,13 +52,13 @@ public class StatelessClientTest { { statelessMap = ChronicleMapBuilder.of(Integer .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8086)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); } // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replicators((byte) 2, TcpReplicationConfig.of(8086)).create(); } @@ -82,7 +84,7 @@ public class StatelessClientTest { } - @Test(timeout = 5000) + @Test public void testBufferOverFlowPutAllAndValues() throws IOException, InterruptedException { final ChronicleMap serverMap; @@ -93,13 +95,13 @@ public class StatelessClientTest { { statelessMap = ChronicleMapBuilder.of(Integer .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8086)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); } // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replicators((byte) 2, TcpReplicationConfig.of(8086)).create(); } @@ -123,7 +125,7 @@ public class StatelessClientTest { } - @Test(timeout = 5000) + @Test public void testBufferOverFlowPutAllAndKeySet() throws IOException, InterruptedException { final ChronicleMap serverMap; @@ -134,13 +136,13 @@ public class StatelessClientTest { { statelessMap = ChronicleMapBuilder.of(Integer .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8086)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); } // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replicators((byte) 2, TcpReplicationConfig.of(8086)).create(); } @@ -164,7 +166,7 @@ public class StatelessClientTest { } - @Test(timeout = 5000) + @Test public void test() throws IOException, InterruptedException { final ChronicleMap serverMap; @@ -173,7 +175,7 @@ public class StatelessClientTest { // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replicators((byte) 2, TcpReplicationConfig.of(8086)).create(); serverMap.put(10, "EXAMPLE-10"); } @@ -183,7 +185,7 @@ public class StatelessClientTest { { statelessMap = ChronicleMapBuilder.of(Integer .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8086)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); Assert.assertEquals("EXAMPLE-10", statelessMap.get(10)); Assert.assertEquals(1, statelessMap.size()); @@ -195,7 +197,7 @@ public class StatelessClientTest { } - @Test(timeout = 5000) + @Test public void testClientCreatedBeforeServer() throws IOException, InterruptedException { final ChronicleMap serverMap; @@ -206,13 +208,13 @@ public class StatelessClientTest { { statelessMap = ChronicleMapBuilder.of(Integer .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8086)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); } // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replicators((byte) 2, TcpReplicationConfig.of(8086)).create(); serverMap.put(10, "EXAMPLE-10"); } @@ -227,7 +229,7 @@ public class StatelessClientTest { } - @Test(timeout = 5000) + @Test public void testBufferOverFlowPutAll() throws IOException, InterruptedException { final ChronicleMap serverMap; @@ -238,13 +240,13 @@ public class StatelessClientTest { { statelessMap = ChronicleMapBuilder.of(Integer .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8086)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); } // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replicators((byte) 2, TcpReplicationConfig.of(8086)).create(); } @@ -268,7 +270,7 @@ public class StatelessClientTest { } - @Test(timeout = 5000) + @Test public void testBufferOverFlowPutAllWherePutReturnsNull() throws IOException, InterruptedException { @@ -281,13 +283,13 @@ public class StatelessClientTest { statelessMap = ChronicleMapBuilder.of(Integer .class, CharSequence.class) .putReturnsNull(true) - .statelessClient(new InetSocketAddress("localhost", 8086)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); } // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)) + .replicators((byte) 2, TcpReplicationConfig.of(8086)) .putReturnsNull(true) .create(); } @@ -313,7 +315,7 @@ public class StatelessClientTest { } - @Test(timeout = 5000) + @Test public void testPutWherePutReturnsNull() throws IOException, InterruptedException { @@ -327,13 +329,13 @@ public class StatelessClientTest { .class, CharSequence.class) .putReturnsNull(true) .removeReturnsNull(true) - .statelessClient(new InetSocketAddress("localhost", 8086)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); } // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)) + .replicators((byte) 2, TcpReplicationConfig.of(8086)) .putReturnsNull(true) .removeReturnsNull(true) .create(); @@ -353,7 +355,7 @@ public class StatelessClientTest { - @Test(timeout = 5000) + @Test public void testRemoveWhereRemoveReturnsNull() throws IOException, InterruptedException { @@ -367,13 +369,13 @@ public class StatelessClientTest { .class, CharSequence.class) .putReturnsNull(true) .removeReturnsNull(true) - .statelessClient(new InetSocketAddress("localhost", 8086)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); } // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)) + .replicators((byte) 2, TcpReplicationConfig.of(8086)) .putReturnsNull(true) .removeReturnsNull(true) .create(); @@ -395,7 +397,7 @@ public class StatelessClientTest { } - @Test(timeout = 5000) + @Test public void testEquals() throws IOException, InterruptedException { final ChronicleMap serverMap1; @@ -407,18 +409,18 @@ public class StatelessClientTest { { statelessMap1 = ChronicleMapBuilder.of(Integer .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8086)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8086))).create(); statelessMap2 = ChronicleMapBuilder.of(Integer .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8077)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", 8077))).create(); } // server { serverMap1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8086)).create(); + .replicators((byte) 2, TcpReplicationConfig.of(8086)).create(); serverMap2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8077)).create(); + .replicators((byte) 2, TcpReplicationConfig.of(8077)).create(); } diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java index f59a03c59..2eec55000 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java @@ -53,6 +53,18 @@ public class TCPSocketReplication4WayMapTest { } + + static ChronicleMap newTcpSocketShmIntValueString( + final byte identifier, + final int serverPort, + final InetSocketAddress... endpoints) throws IOException { + return newTcpSocketShmBuilder(IntValue.class, CharSequence.class, + identifier, serverPort, endpoints) + .entries(20000L) + .keyMarshaller(ByteableIntValueMarshaller.INSTANCE).create(); + } + + @Before public void setup() throws IOException { diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java index 5dae68d57..7eb7f7d7c 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java @@ -51,7 +51,7 @@ public class TCPSocketReplicationBootStrapTests { newTcpSocketShmBuilder(Integer.class, CharSequence.class, (byte) 2, 8076, new InetSocketAddress("localhost", 8079)); final ChronicleMap map2a = - map2aBuilder.createPersistedTo(getPersistenceFile()); + map2aBuilder.file(getPersistenceFile()).create(); map2a.put(10, "EXAMPLE-10"); // this will be the last time that map1 go an update from map2 long lastModificationTime; @@ -69,14 +69,16 @@ public class TCPSocketReplicationBootStrapTests { // restart map 2 but don't doConnect it to map one final ChronicleMap map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .createPersistedTo(map2File); + .file(map2File) + .create(); // add data into it map2b.put(11, "ADDED WHEN DISCONNECTED TO MAP1"); map2b.close(); } // now restart map2a and doConnect it to map1, map1 should bootstrap the missing entry - map2 = map2aBuilder.createPersistedTo(map2File); + map2aBuilder.file(map2File); + map2 = map2aBuilder.create(); // add data into it waitTillEqual(5000); @@ -94,7 +96,7 @@ public class TCPSocketReplicationBootStrapTests { final ReplicatedChronicleMap map2a = (ReplicatedChronicleMap) - map2aBuilder.createPersistedTo(getPersistenceFile()); + map2aBuilder.file(getPersistenceFile()).create(); map2a.put(10, "EXAMPLE-10"); // this will be the last time that map1 go an update from map2 @@ -111,16 +113,18 @@ public class TCPSocketReplicationBootStrapTests { { // restart map 2 but don't doConnect it to map one + ChronicleMapBuilder.of(Integer.class, CharSequence.class).file(map2File); final ChronicleMap map2b = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .createPersistedTo(map2File); + ChronicleMapBuilder.of(Integer.class, CharSequence.class).file(map2File). + create(); // add data into it map2b.put(11, "ADDED WHEN DISCONNECTED TO MAP1"); map2b.close(); } // now restart map2a and doConnect it to map1, map1 should bootstrap the missing entry - map2 = map2aBuilder.createPersistedTo(map2File); + map2aBuilder.file(map2File); + map2 = map2aBuilder.create(); // add data into it waitTillEqual(20000); diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java index edd1aa3c4..982d5ca78 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java @@ -34,6 +34,7 @@ import java.util.Random; import static net.openhft.chronicle.map.Builder.getPersistenceFile; import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder; +import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmIntValueString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -56,14 +57,9 @@ public class TCPSocketReplicationIntValueTest { ((Byteable) value).bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0); map1Builder = newTcpSocketShmBuilder(IntValue.class, CharSequence.class, (byte) 1, 8076, new InetSocketAddress("localhost", 8077)); - map1Builder.keyMarshaller(ByteableIntValueMarshaller.INSTANCE); - map1 = map1Builder.keyMarshaller(ByteableIntValueMarshaller.INSTANCE) - .createPersistedTo(getPersistenceFile()); - map2 = newTcpSocketShmBuilder(IntValue.class, CharSequence.class, - (byte) 2, 8077) - .entries(20000L) - .keyMarshaller(ByteableIntValueMarshaller.INSTANCE) - .create(); + map1Builder.keyMarshaller(ByteableIntValueMarshaller.INSTANCE).file(getPersistenceFile()); + map1 = map1Builder.keyMarshaller(ByteableIntValueMarshaller.INSTANCE).create(); + map2 = newTcpSocketShmIntValueString((byte) 2, 8077); } diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java index fac6062dd..7ba19246e 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java @@ -50,7 +50,8 @@ public class TCPSocketReplicationTest { ChronicleMapBuilder map1Builder = newTcpSocketShmBuilder(Integer.class, CharSequence.class, (byte) 1, 8076, new InetSocketAddress("localhost", 8077)); - map1 = map1Builder.createPersistedTo(getPersistenceFile()); + map1Builder.file(getPersistenceFile()); + map1 = map1Builder.create(); map2 = newTcpSocketShmIntString((byte) 2, 8077); } diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java index 23703da79..4a2f2a52e 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.TcpReplicationConfig; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -26,7 +26,6 @@ import org.junit.Test; import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; -import java.util.Arrays; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -48,11 +47,11 @@ public class TCPSocketReplicationTest3wayPutReturnsNull { final byte identifier, final int serverPort, final InetSocketAddress... endpoints) throws IOException { - TcpConfig tcpConfig = TcpConfig.unknownTopology(serverPort, Arrays.asList(endpoints)); + TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(serverPort, endpoints); return (T) ChronicleMapBuilder.of(Integer.class, CharSequence.class) .putReturnsNull(true) - .replication(identifier, tcpConfig).create(); + .replicators(identifier, tcpConfig).create(); } diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTestPostConnection.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTestPostConnection.java index 4fcaf22c9..c2656ddd5 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTestPostConnection.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTestPostConnection.java @@ -79,8 +79,9 @@ public class TCPSocketReplicationTestPostConnection { ChronicleMapBuilder map2aBuilder = newTcpSocketShmBuilder(Integer.class, CharSequence.class, (byte) 2, 8077, new InetSocketAddress("localhost", 8076)); + map2aBuilder.file(Builder.getPersistenceFile()); final ChronicleMap map2a = - map2aBuilder.createPersistedTo(Builder.getPersistenceFile()); + map2aBuilder.create(); map1 = TCPSocketReplication4WayMapTest.newTcpSocketShmIntString((byte) 1, 8076); Thread.sleep(1); @@ -93,7 +94,8 @@ public class TCPSocketReplicationTestPostConnection { map1.put(6, "EXAMPLE-1"); // recreate map2 with new unique file - map2 = map2aBuilder.createPersistedTo(Builder.getPersistenceFile()); + map2aBuilder.file(Builder.getPersistenceFile()); + map2 = map2aBuilder.create(); // allow time for the recompilation to resolve diff --git a/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTests.java b/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTests.java index e26b077bd..f35862472 100644 --- a/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTests.java +++ b/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTests.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TimeProvider; +import net.openhft.chronicle.hash.TimeProvider; import net.openhft.chronicle.map.jrs166.JSR166TestCase; import org.junit.Test; import org.mockito.Mockito; @@ -49,10 +49,12 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALatePut() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10).replicators((byte) 1); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .timeProvider(timeProvider) - .createReplicated((byte) 1); + .replicators((byte) 1) + .timeProvider(timeProvider).create(); current(timeProvider); @@ -78,10 +80,11 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALatePutIfAbsent() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10).replicators((byte) 1); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .timeProvider(timeProvider) - .createReplicated((byte) 1); + .replicators((byte) 1) + .timeProvider(timeProvider).create(); current(timeProvider); @@ -108,10 +111,11 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALateReplace() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10).replicators((byte) 1); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .timeProvider(timeProvider) - .createReplicated((byte) 1); + .replicators((byte) 1) + .timeProvider(timeProvider).create(); current(timeProvider); @@ -141,10 +145,14 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALateReplaceWithValue() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10).replicators((byte) 1); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .replicators((byte) 1) + .timeProvider(timeProvider).file(getPersistenceFile()); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .timeProvider(timeProvider) - .createReplicated((byte) 1); + .replicators((byte) 1) + .timeProvider(timeProvider).create(); current(timeProvider); @@ -171,10 +179,13 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALateRemoveWithValue() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class).replicators((byte) 1); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .replicators((byte) 1) + .timeProvider(timeProvider).file(getPersistenceFile()); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .timeProvider(timeProvider) - .createReplicated((byte) 1); + .replicators((byte) 1) + .timeProvider(timeProvider).create(); current(timeProvider); @@ -200,10 +211,13 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALateRemove() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class).replicators((byte) 1); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .replicators((byte) 1) + .timeProvider(timeProvider).file(getPersistenceFile()); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .timeProvider(timeProvider) - .createReplicated((byte) 1); + .replicators((byte) 1) + .timeProvider(timeProvider).create(); current(timeProvider); @@ -229,10 +243,13 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreWithRemoteRemove() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class).replicators((byte) 1); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .replicators((byte) 1) + .timeProvider(timeProvider).file(getPersistenceFile()); ReplicatedChronicleMap map = (ReplicatedChronicleMap) ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .timeProvider(timeProvider) - .createReplicated((byte) 1); + .replicators((byte) 1) + .timeProvider(timeProvider).create(); current(timeProvider); @@ -259,10 +276,13 @@ public class TimeBasedReplicationTests extends JSR166TestCase { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class).replicators((byte) 1); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .replicators((byte) 1) + .timeProvider(timeProvider).file(getPersistenceFile()); ReplicatedChronicleMap map = (ReplicatedChronicleMap) ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .timeProvider(timeProvider) - .createReplicated((byte) 1); + .replicators((byte) 1) + .timeProvider(timeProvider).create(); current(timeProvider); @@ -289,10 +309,16 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testRemoveFollowedByLatePut() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10) + .replicators((byte) 1) + .entries(10) + .timeProvider(timeProvider).file(getPersistenceFile()); ReplicatedChronicleMap map = (ReplicatedChronicleMap) ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) .entries(10) - .timeProvider(timeProvider) - .createReplicated((byte) 1); + .replicators((byte) 1) + .entries(10) + .timeProvider(timeProvider).create(); current(timeProvider); @@ -317,10 +343,16 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testPutRemovePut() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10).replicators((byte) 1); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10) + .entries(10) + .timeProvider(timeProvider).file(getPersistenceFile()); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) .entries(10) - .timeProvider(timeProvider) - .createReplicated((byte) 1); + .entries(10) + .timeProvider(timeProvider).create(); current(timeProvider); diff --git a/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java b/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java index 8bb712717..6d64e8ae2 100644 --- a/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java +++ b/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.TcpReplicationConfig; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -30,6 +30,7 @@ import java.net.InetSocketAddress; import static java.util.concurrent.TimeUnit.SECONDS; import static net.openhft.chronicle.map.Builder.getPersistenceFile; +import static net.openhft.chronicle.hash.TcpReplicationConfig.of; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -45,20 +46,19 @@ public class TwoMapOnDifferentServers { @Before public void setup() throws IOException { - final TcpConfig tcpConfig = - TcpConfig.forSendingNode(8076, new InetSocketAddress("localhost", 8077)) + final TcpReplicationConfig tcpConfig = + of(8076, new InetSocketAddress("localhost", 8077)) .heartBeatInterval(1, SECONDS); File persistenceFile = getPersistenceFile(); map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000) - .replication((byte) 1, tcpConfig).create(); + .replicators((byte) 1, tcpConfig).create(); map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(8077) - .heartBeatInterval(1, SECONDS)).create(); + .replicators((byte) 2, of(8077).heartBeatInterval(1, SECONDS)).create(); } @After diff --git a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java index 1dc54598a..22887c0d8 100644 --- a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java @@ -18,8 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.SimpleReplication; -import net.openhft.chronicle.hash.replication.UdpConfig; +import net.openhft.chronicle.hash.UdpReplicationConfig; import org.junit.After; import org.junit.Before; @@ -41,13 +40,12 @@ public class UDPSocketReplicationTest { final int identifier, final int udpPort) throws IOException { - UdpConfig udpConfig = UdpConfig + UdpReplicationConfig udpConfig = UdpReplicationConfig .simple(Inet4Address.getByName("255.255.255.255"), udpPort); return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .replication(SimpleReplication.builder().udpTransport(udpConfig) - .create((byte) identifier)).create(); + .replicators((byte) identifier, udpConfig).create(); } @Before diff --git a/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java b/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java index dbff0417d..404c74741 100644 --- a/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java +++ b/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java @@ -1,7 +1,8 @@ package net.openhft.chronicle.map.example; -import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.ChannelProvider; +import net.openhft.chronicle.hash.ChannelProviderBuilder; +import net.openhft.chronicle.hash.TcpReplicationConfig; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import org.junit.Assert; @@ -10,7 +11,6 @@ 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; /** @@ -54,8 +54,8 @@ public class ChannelExample { private ChronicleMap favoriteColourServer1; - private ReplicationHub hubOnServer1; - private ReplicationHub hubOnServer2; + private ChannelProvider channelProviderServer1; + private ChannelProvider channelProviderServer2; @Test @@ -63,34 +63,32 @@ public class ChannelExample { // server 1 with identifier = 1 { - ChronicleMapBuilder smallStringToStringMapBuilder = - ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(1000); - byte identifier = (byte) 1; - TcpConfig tcpConfig = TcpConfig - .forSendingNode(8086, new InetSocketAddress("localhost", 8087)) - .heartBeatInterval(1, SECONDS); + TcpReplicationConfig tcpConfig = TcpReplicationConfig + .of(8086, new InetSocketAddress("localhost", 8087)) + .heartBeatInterval(1, java.util.concurrent.TimeUnit.SECONDS); - hubOnServer1 = ReplicationHub.builder() - .tcpTransportAndNetwork(tcpConfig) - .create(identifier); + channelProviderServer1 = new ChannelProviderBuilder() + .replicators(identifier, tcpConfig) + .create(); // this demotes favoriteColour short channel1 = (short) 1; - favoriteColourServer1 = smallStringToStringMapBuilder.instance() - .replicatedViaChannel(hubOnServer1.createChannel(channel1)).create(); + favoriteColourServer1 = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000) + .channel(channelProviderServer1.createChannel(channel1)).create(); favoriteColourServer1.put("peter", "green"); // this demotes favoriteComputer short channel2 = (short) 2; - favoriteComputerServer1 = smallStringToStringMapBuilder.instance() - .replicatedViaChannel(hubOnServer1.createChannel(channel2)).create(); + favoriteComputerServer1 = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000) + .channel(channelProviderServer1.createChannel(channel2)).create(); favoriteComputerServer1.put("peter", "dell"); @@ -98,25 +96,23 @@ public class ChannelExample { // server 2 with identifier = 2 { - ChronicleMapBuilder smallStringToStringMapBuilder = - ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(1000); byte identifier = (byte) 2; - TcpConfig tcpConfig = - TcpConfig.forReceivingOnlyNode(8087).heartBeatInterval(1, SECONDS); + TcpReplicationConfig tcpConfig = + TcpReplicationConfig.of(8087).heartBeatInterval(1, java.util.concurrent.TimeUnit.SECONDS); - hubOnServer2 = ReplicationHub.builder() - .tcpTransportAndNetwork(tcpConfig) - .create(identifier); + channelProviderServer2 = new ChannelProviderBuilder() + .replicators(identifier, tcpConfig) + .create(); // this demotes favoriteColour short channel1 = (short) 1; - favoriteColourServer2 = smallStringToStringMapBuilder.instance() - .replicatedViaChannel(hubOnServer2.createChannel(channel1)).create(); + favoriteColourServer2 = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000) + .channel(channelProviderServer2.createChannel(channel1)).create(); favoriteColourServer2.put("rob", "blue"); @@ -124,8 +120,9 @@ public class ChannelExample { // this demotes favoriteComputer short channel2 = (short) 2; - favoriteComputerServer2 = smallStringToStringMapBuilder.instance() - .replicatedViaChannel(hubOnServer2.createChannel(channel2)).create(); + favoriteComputerServer2 = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000) + .channel(channelProviderServer2.createChannel(channel2)).create(); favoriteComputerServer2.put("rob", "mac"); favoriteComputerServer2.put("daniel", "mac"); diff --git a/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java b/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java index 1a8d3b140..562ad1a71 100644 --- a/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java +++ b/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java @@ -30,6 +30,9 @@ public class DistributedSequenceMain { public static void main(String... ignored) throws IOException { Class longValueClass = DataValueClasses.directClassFor(LongValue.class); + ChronicleMapBuilder.of(String.class, longValueClass) + .entries(128) + .actualSegments(1).file(new File("/tmp/counters")); ChronicleMap map = ChronicleMapBuilder.of(String.class, longValueClass) .entries(128) diff --git a/src/test/java/net/openhft/chronicle/map/fromdocs/OpenJDKAndHashMapExamplesTest.java b/src/test/java/net/openhft/chronicle/map/fromdocs/OpenJDKAndHashMapExamplesTest.java index e33723d2d..35dc2aa58 100644 --- a/src/test/java/net/openhft/chronicle/map/fromdocs/OpenJDKAndHashMapExamplesTest.java +++ b/src/test/java/net/openhft/chronicle/map/fromdocs/OpenJDKAndHashMapExamplesTest.java @@ -59,7 +59,8 @@ public class OpenJDKAndHashMapExamplesTest { ChronicleMap chm = OffHeapUpdatableChronicleMapBuilder .of(String.class, BondVOInterface.class) .keySize(10) - .createPersistedTo(file); + .file(file) + .create(); BondVOInterface bondVO = DataValueClasses.newDirectReference(BondVOInterface.class); @@ -79,7 +80,8 @@ public class OpenJDKAndHashMapExamplesTest { ChronicleMap chmB = OffHeapUpdatableChronicleMapBuilder .of(String.class, BondVOInterface.class) .keySize(10) - .createPersistedTo(file); + .file(file) + .create(); // ZERO Copy but creates a new off heap reference each time diff --git a/src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongCASLeft.java b/src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongCASLeft.java index 0d31aaade..df78c1860 100644 --- a/src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongCASLeft.java +++ b/src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongCASLeft.java @@ -90,7 +90,14 @@ public class PingPongCASLeft { static ChronicleMap acquireCHM() throws IOException { // ensure thread ids are globally unique. AffinitySupport.setThreadId(); + + String TMP = System.getProperty("java.io.tmpdir"); + ChronicleMapBuilder.of(String.class, BondVOInterface.class) + + .entries(16) + .entrySize(64).file(new File(TMP + "/BondPortfolioCHM")); return ChronicleMapBuilder.of(String.class, BondVOInterface.class) + .entries(16) .entrySize(64).create(); } diff --git a/src/test/java/net/openhft/chronicle/map/ipc/StateMachineTutorial.java b/src/test/java/net/openhft/chronicle/map/ipc/StateMachineTutorial.java index 7cc518e78..9abbd9dd4 100644 --- a/src/test/java/net/openhft/chronicle/map/ipc/StateMachineTutorial.java +++ b/src/test/java/net/openhft/chronicle/map/ipc/StateMachineTutorial.java @@ -40,7 +40,12 @@ public class StateMachineTutorial { try { File dataFile = new File(System.getProperty("java.io.tmpdir"), "hft-state-machine"); + + ChronicleMapBuilder.of(Integer.class, StateMachineData.class) + + .entries(8).file(dataFile); map = ChronicleMapBuilder.of(Integer.class, StateMachineData.class) + .entries(8).create(); if (args.length > 0) { 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 c3665fa8c..7620fb4a3 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 @@ -49,33 +49,57 @@ public class ChronicleMapTest extends JSR166TestCase { } static ChronicleMap newShmIntString(int size) throws IOException { + + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(size).file(getPersistenceFile()); return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(size).create(); } static ChronicleMap newShmListBoolean(int size) throws IOException { + + ChronicleMapBuilder.of(ArrayList.class, CharSequence.class) + + .entries(size).file(getPersistenceFile()); return ChronicleMapBuilder.of(ArrayList.class, CharSequence.class) + .entries(size).create(); + } static ChronicleMap newShmListBoolean() throws IOException { + + ChronicleMapBuilder.of(ArrayList.class, CharSequence.class).file(getPersistenceFile()); return ChronicleMapBuilder.of(ArrayList.class, CharSequence.class).create(); + } static ChronicleMap newShmStringString(int size) throws IOException { + + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + + .entries(size).file(getPersistenceFile()); return ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(size).create(); + } static ChronicleMap newShmIntString() throws IOException { + + ChronicleMapBuilder.of(Integer.class, CharSequence.class).file(getPersistenceFile()); return ChronicleMapBuilder.of(Integer.class, CharSequence.class).create(); + } static ChronicleMap newShmBiBoolean() throws IOException { + + ChronicleMapBuilder.of(BI.class, Boolean.class).file(getPersistenceFile()); return ChronicleMapBuilder.of(BI.class, Boolean.class).create(); + } /** diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/MapCheck.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/MapCheck.java index b076f8e07..261d4d796 100644 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/MapCheck.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/MapCheck.java @@ -123,6 +123,9 @@ public class MapCheck { static Map newMap() { try { + +// return new ConcurrentHashMap(); + ChronicleMapBuilder.of(Object.class, Object.class).file(getPersistenceFile()); return ChronicleMapBuilder.of(Object.class, Object.class).create(); } catch (Exception e) { diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java index 1f19d83e3..e66eef94e 100644 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map.jrs166.map; -import net.openhft.chronicle.hash.replication.TcpConfig; +import net.openhft.chronicle.hash.TcpReplicationConfig; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import net.openhft.chronicle.map.ReadContext; @@ -33,6 +33,7 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.util.*; +import static net.openhft.chronicle.hash.StatelessBuilder.remoteAddress; import static org.junit.Assert.*; /* @@ -183,13 +184,12 @@ public class StatelessChronicleMapTest extends JSR166TestCase { static ChronicleMap newShmIntString(int port) throws IOException { - final ChronicleMap serverMap = - ChronicleMapBuilder.of(Integer.class, String.class) - .replication((byte) 1, TcpConfig.forReceivingOnlyNode(port)).create(); + final ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, String.class) + .replicators((byte) 1, TcpReplicationConfig.of(port)).create(); final ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer .class, String.class) - .statelessClient(new InetSocketAddress("localhost", port)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", port))).create(); return new SingleCloseMap(statelessMap, statelessMap, serverMap); @@ -199,12 +199,12 @@ public class StatelessChronicleMapTest extends JSR166TestCase { IOException { final ChronicleMap serverMap = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .replication((byte) 2, TcpConfig.forReceivingOnlyNode(port)).create(); + .replicators((byte) 2, TcpReplicationConfig.of(port)).create(); final ChronicleMap statelessMap = ChronicleMapBuilder.of(CharSequence .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", port)).create(); + .stateless(remoteAddress(new InetSocketAddress("localhost", port))).create(); return new SingleCloseMap(statelessMap, statelessMap, serverMap); } diff --git a/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java b/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java index cecae189c..4997e3c36 100644 --- a/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java +++ b/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java @@ -138,7 +138,8 @@ public class ProcessInstanceLimiter implements Runnable { ChronicleMapBuilder.of(String.class, Data.class); builder.entries(1000); builder.entrySize(1024); - this.theSharedMap = builder.createPersistedTo(new File(sharedMapPath)); + builder.file(new File(sharedMapPath)); + this.theSharedMap = builder.create(); Thread t = new Thread(this, "ProcessInstanceLimiter updater"); t.setDaemon(true); t.start(); diff --git a/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java b/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java index 88709f59f..efb1d7868 100644 --- a/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java +++ b/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java @@ -18,7 +18,6 @@ package net.openhft.chronicle.set; -import net.openhft.chronicle.hash.replication.TcpConfig; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -26,11 +25,10 @@ import org.junit.Test; import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; -import java.util.Arrays; -import static java.util.Arrays.asList; import static java.util.concurrent.TimeUnit.SECONDS; import static net.openhft.chronicle.map.Builder.getPersistenceFile; +import static net.openhft.chronicle.hash.TcpReplicationConfig.of; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -53,9 +51,10 @@ public class SetForTCPSocketReplicationTest3way { final int serverPort, final InetSocketAddress... endpoints) throws IOException { return (T) ChronicleSetBuilder.of(Integer.class) - .replication(identifier, TcpConfig.unknownTopology(serverPort, asList(endpoints)) + .file(getPersistenceFile()) + .replicators(identifier, of(serverPort, endpoints) .heartBeatInterval(1L, SECONDS)) - .createPersistedTo(getPersistenceFile()); + .create(); }