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

reverted cff7c5bea8

Browse Source
This commit is contained in:
Rob Austin
2014-11-04 14:44:59 +00:00
parent a0c89935f5
commit 9dec1ed4db
64 changed files with 700 additions and 865 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
+92 -60
View File
@@ -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<K, C extends ChronicleHash,
B metaDataBytes(int metaDataBytes);
/**
* Configures replication of the hash containers, created by this builder (via independent
* replicators). See <a href="https://github.com/OpenHFT/Chronicle-Map#tcp--udp-replication">
* the section about replication in ChronicleMap manual</a> for more information.
*
* <p>Another way to establish replication is {@link #channel(ChannelProvider.ChronicleChannel)}
* method.
*
* <p>By default, hash containers, created by this builder doesn't replicate their data.
*
* <p>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 <a href="https://github.com/OpenHFT/Chronicle-Map#channels-and-channelprovider">the
* section about Channels and ChannelProvider in ChronicleMap manual</a> for more information.
*
* <p>Another way to establish replication if {@link #replicators(byte, ReplicationConfig[])}
* method.
*
* <p>By default, hash containers, created by this builder doesn't replicate their data.
*
* <p>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.
*
* <p>By default replication is not configured, so this method is useful <i>only</i> 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<K, C extends ChronicleHash,
*
* @param timeProvider a new time provider for replication needs
* @return this builder back
* @see #replication(SimpleReplication)
* @see #replicators(byte, ReplicationConfig[])
* @see #channel(ChannelProvider.ChronicleChannel)
*/
B timeProvider(TimeProvider timeProvider);
@@ -329,37 +387,38 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
B immutableKeys();
/**
* Configures replication of the hash containers, created by this builder.
* See <a href="https://github.com/OpenHFT/Chronicle-Map#tcp--udp-replication">
* the section about replication in ChronicleMap manual</a> 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
*
* <p>By default, hash containers, created by this builder doesn't replicate their data.
*
* <p>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<C> statelessClient(InetSocketAddress remoteAddress);
ChronicleHashInstanceConfig<C> 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 ChronicleHashBuilder<K, C extends ChronicleHash,
* object is collected during GC, or on JVM shutdown the off-heap memory used by the returned
* container is freed.
*
* <p>This method is a shortcut for {@code instance().create()}.
*
* @return a new off-heap hash container
* @throws IOException if any IO error relates to off-heap memory allocation 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.
*
* <p>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.
*
* <p>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.
*
* <p>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;
}
src/main/java/net/openhft/chronicle/hash/serialization/BytesInterop.java
+4 -19
View File
@@ -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.
*
* <p>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 <E> type of marshalled objects
*/
public interface BytesInterop<E> extends BytesWriter<E> {
/**
* 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);
src/main/java/net/openhft/chronicle/hash/serialization/BytesReader.java
+1 -74
View File
@@ -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.
*
* <p>Implementation example:<pre><code>
* class LongPair { long first, second; }
*
* enum LongPairArrayReader implements BytesReader&lt;LongPair[]&gt; {
* INSTANCE;
*
* &#064;Override
* public LongPair[] read(Bytes bytes, long size) {
* return read(bytes, size, null);
* }
*
* &#064;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;
* }
* }</code></pre>
*
* @param <E> the type of the object demarshalled
* @see BytesWriter
*/
public interface BytesReader<E> extends Serializable {
/**
* Reads and returns the object from {@code [position(), position() + size]} bytes of the given
* {@link Bytes}.
*
* <p>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);
}
src/main/java/net/openhft/chronicle/hash/serialization/BytesWriter.java
-53
View File
@@ -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.
*
* <p>Implementation example: <pre><code>
* class LongPair { long first, second; }
*
* enum LongPairArrayWriter implements BytesWriter&lt;LongPair[]&gt; {
* INSTANCE;
*
* &#064;Override
* public long size(LongPair[] longPairs) {
* return longPairs.length * 16L;
* }
*
* &#064;Override
* public void write(Bytes bytes, LongPair[] longPairs) {
* for (LongPair pair : longPairs) {
* bytes.writeLong(pair.first);
* bytes.writeLong(pair.second);
* }
* }
* }</code></pre>
*
* @param <E> the type of the object marshalled
* @see BytesReader
* @see BytesInterop
*/
public interface BytesWriter<E> 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.
*
* <p>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);
}
src/main/java/net/openhft/chronicle/hash/serialization/impl/ByteArrayMarshaller.java
+5 -5
View File
@@ -59,10 +59,10 @@ public enum ByteArrayMarshaller implements BytesInterop<byte[]>, BytesReader<byt
}
@Override
public byte[] read(Bytes bytes, long size, byte[] toReuse) {
if (toReuse == null || size != toReuse.length)
toReuse = new byte[(int) size];
bytes.read(toReuse);
return toReuse;
public byte[] read(Bytes bytes, long size, byte[] ba) {
if (ba == null || size != ba.length)
ba = new byte[(int) size];
bytes.read(ba);
return ba;
}
}
src/main/java/net/openhft/chronicle/hash/serialization/impl/ByteableMarshaller.java
+7 -10
View File
@@ -18,10 +18,8 @@
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 net.openhft.lang.io.MultiStoreBytes;
import net.openhft.lang.io.NativeBytes;
@@ -30,8 +28,7 @@ import net.openhft.lang.io.serialization.impl.AllocateInstanceObjectFactory;
import net.openhft.lang.model.Byteable;
import org.jetbrains.annotations.NotNull;
public class ByteableMarshaller<E extends Byteable>
implements BytesInterop<E>, BytesReader<E>, SizeMarshaller {
public class ByteableMarshaller<E extends Byteable> implements AgileBytesMarshaller<E> {
private static final long serialVersionUID = 0L;
public static <E extends Byteable> ByteableMarshaller<E> of(@NotNull Class<E> eClass) {
@@ -117,13 +114,13 @@ public class ByteableMarshaller<E extends Byteable>
}
@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);
}
src/main/java/net/openhft/chronicle/hash/serialization/impl/CharArrayMarshaller.java
+5 -5
View File
@@ -49,11 +49,11 @@ public enum CharArrayMarshaller implements BytesInterop<char[]>, BytesReader<cha
}
@Override
public char[] read(Bytes bytes, long size, char[] toReuse) {
if (toReuse == null || toReuse.length != size)
toReuse = new char[(int) size];
bytes.readFully(toReuse);
return toReuse;
public char[] read(Bytes bytes, long size, char[] chars) {
if (chars == null || chars.length != size)
chars = new char[(int) size];
bytes.readFully(chars);
return chars;
}
@Override
src/main/java/net/openhft/chronicle/hash/serialization/impl/CharSequenceReader.java
+9 -9
View File
@@ -78,14 +78,14 @@ public final class CharSequenceReader<S extends CharSequence>
}
@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<S extends CharSequence>
} catch (IOException e) {
throw new IllegalStateException(e);
}
if (appendable == toReuse)
return toReuse;
if (appendable == s)
return s;
return interner.intern(sb);
}
src/main/java/net/openhft/chronicle/hash/serialization/impl/DoubleMarshaller.java
+3 -5
View File
@@ -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<Double>, BytesReader<Double>, SizeMarshaller {
public enum DoubleMarshaller implements AgileBytesMarshaller<Double> {
INSTANCE;
@Override
@@ -70,7 +68,7 @@ public enum DoubleMarshaller implements BytesInterop<Double>, BytesReader<Double
}
@Override
public Double read(Bytes bytes, long size, Double toReuse) {
public Double read(Bytes bytes, long size, Double e) {
return read(bytes, size);
}
}
src/main/java/net/openhft/chronicle/hash/serialization/impl/IntegerMarshaller.java
+4 -4
View File
@@ -18,11 +18,11 @@
package net.openhft.chronicle.hash.serialization.impl;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.hash.serialization.AgileBytesMarshaller;
import net.openhft.chronicle.hash.serialization.Hasher;
import net.openhft.lang.io.Bytes;
public enum IntegerMarshaller
implements BytesInterop<Integer>, BytesReader<Integer>, SizeMarshaller {
public enum IntegerMarshaller implements AgileBytesMarshaller<Integer> {
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();
}
}
src/main/java/net/openhft/chronicle/hash/serialization/impl/LongMarshaller.java
+3 -6
View File
@@ -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<Long>, BytesReader<Long>, SizeMarshaller {
public enum LongMarshaller implements AgileBytesMarshaller<Long> {
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();
}
}
src/main/java/net/openhft/chronicle/hash/serialization/impl/VoidMarshaller.java
+3 -6
View File
@@ -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<Void>, BytesReader<Void>, SizeMarshaller {
public enum VoidMarshaller implements AgileBytesMarshaller<Void> {
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;
}
src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+1 -1
View File
@@ -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;
src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java
+109 -115
View File
@@ -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<K, V,
SerializationBuilder<K> keyBuilder;
SerializationBuilder<V> valueBuilder;
private Map<Class<? extends Replicator>, Replicator> replicators = new HashMap<Class<? extends Replicator>, Replicator>();
// used when configuring the number of segments.
private int minSegments = -1;
private int actualSegments = -1;
@@ -135,8 +133,10 @@ public abstract class AbstractChronicleMapBuilder<K, V,
private V defaultValue = null;
private DefaultValueProvider<K, V> defaultValueProvider = NullValueProvider.INSTANCE;
private PrepareValueBytes<K, V> prepareValueBytes = null;
private byte identifier = -1;
private SimpleReplication simpleReplication = null;
private StatelessBuilder statelessBuilder;
private File file;
AbstractChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
keyBuilder = new SerializationBuilder<K>(keyClass, SerializationBuilder.Role.KEY);
@@ -283,14 +283,14 @@ public abstract class AbstractChronicleMapBuilder<K, V,
return self();
}
int entrySize(boolean replicated) {
int entrySize() {
if (entrySize > 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<K, V,
return toString().hashCode();
}
public B replicators(byte identifier, ReplicationConfig... replicationConfigs) {
this.identifier = identifier;
this.replicators.clear();
for (ReplicationConfig replicationConfig : replicationConfigs) {
Replicator replicator;
if (replicationConfig instanceof TcpReplicationConfig) {
replicator = Replicators.tcp((TcpReplicationConfig) replicationConfig);
} else if (replicationConfig instanceof UdpReplicationConfig) {
replicator = Replicators.udp((UdpReplicationConfig) replicationConfig);
} else
throw new UnsupportedOperationException();
this.replicators.put(replicator.getClass(), replicator);
}
return self();
}
public B channel(ChannelProvider.ChronicleChannel chronicleChannel) {
this.identifier = chronicleChannel.identifier();
this.replicators.clear();
replicators.put(chronicleChannel.getClass(), chronicleChannel);
return self();
}
@Override
public B disableReplication() {
identifier = -1;
replicators.clear();
return self();
}
@Override
public B timeProvider(TimeProvider timeProvider) {
this.timeProvider = timeProvider;
@@ -619,9 +656,15 @@ public abstract class AbstractChronicleMapBuilder<K, V,
}
@Override
public StatelessClientConfig<ChronicleMap<K, V>> 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<K, V,
/**
* Non-public because should be called only after {@link #preMapConstruction(boolean)}
* Non-public because should be called only after {@link #preMapConstruction()}
*/
DefaultValueProvider<K, V> defaultValueProvider() {
if (defaultValueProvider != null)
@@ -723,43 +766,7 @@ public abstract class AbstractChronicleMapBuilder<K, V,
return new ConstantValueProvider<K, V>(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<ChronicleMap<K, V>> instance() {
return new InstanceConfig<>(this.clone(), simpleReplication, null, null);
}
@Override
public ChronicleMap<K, V> createPersistedTo(File file) throws IOException {
return createWithFile(file, simpleReplication, null);
}
@Override
public ChronicleMap<K, V> create() throws IOException {
return createWithoutFile(simpleReplication, null);
}
ChronicleMap<K, V> create(InstanceConfig<K, V> ib) throws IOException {
if (ib.file != null) {
return createWithFile(ib.file, ib.simpleReplication, ib.channel);
} else {
return createWithoutFile(ib.simpleReplication, ib.channel);
}
}
ChronicleMap<K, V> createWithFile(File file, SimpleReplication simpleReplication,
ReplicationChannel channel) throws IOException {
public ChronicleMap<K, V> createWithFile(File file) throws IOException {
for (int i = 0; i < 10; i++) {
if (file.exists() && file.length() > 0) {
FileInputStream fis = new FileInputStream(file);
@@ -769,7 +776,7 @@ public abstract class AbstractChronicleMapBuilder<K, V,
(VanillaChronicleMap<K, ?, ?, V, ?, ?>) 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<K, V,
if (!file.exists())
throw new FileNotFoundException("Unable to create " + file);
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap(simpleReplication, channel);
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap();
FileOutputStream fos = new FileOutputStream(file);
ObjectOutputStream oos = new ObjectOutputStream(fos);
@@ -802,66 +809,54 @@ public abstract class AbstractChronicleMapBuilder<K, V,
oos.close();
}
return establishReplication(map, simpleReplication, channel);
return establishReplication(map);
}
/**
* Only for testing.
*
* @param identifier id
* @return map
*/
ChronicleMap<K, V> createReplicated(byte identifier) throws IOException {
VanillaChronicleMap<K, ?, ?, V, ?, ?> map =
new ReplicatedChronicleMap<K, Object, MetaBytesInterop<K, Object>,
V, Object, MetaBytesWriter<V, Object>>(this, identifier);
@Override
public ChronicleMap<K, V> create() throws IOException {
if (statelessBuilder != null)
return createStatelessMap(statelessBuilder);
return (file != null) ? createWithFile(file) : createWithoutFile();
}
private ChronicleMap<K, V> createWithoutFile() throws IOException {
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap();
BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE,
map.sizeInBytes(), true);
map.createMappedStoreAndSegments(bytesStore);
return map;
return establishReplication(map);
}
private ChronicleMap<K, V> createWithoutFile(
SimpleReplication simpleReplication, ReplicationChannel channel) throws IOException {
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap(simpleReplication, channel);
BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE,
map.sizeInBytes(), true);
map.createMappedStoreAndSegments(bytesStore);
return establishReplication(map, simpleReplication, channel);
}
ChronicleMap<K, V> createStatelessMap(StatelessMapConfig<K, V> statelessBuilder)
throws IOException {
preMapConstruction(false);
private ChronicleMap<K, V> createStatelessMap(StatelessBuilder statelessBuilder) throws IOException {
preMapConstruction();
return new StatelessChronicleMap<K, V>(
statelessBuilder,
this);
}
private VanillaChronicleMap<K, ?, ?, V, ?, ?> 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<K, ?, ?, V, ?, ?> newMap() throws IOException {
preMapConstruction();
if (useReplication()) {
return new ReplicatedChronicleMap<K, Object, MetaBytesInterop<K, Object>,
V, Object, MetaBytesWriter<V, Object>>(this, identifier);
V, Object, MetaBytesWriter<V, Object>>(this);
} else {
return new VanillaChronicleMap<K, Object, MetaBytesInterop<K, Object>,
V, Object, MetaBytesWriter<V, Object>>(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<K, V,
valueBuilder.constantSizeBySample(sampleValue);
}
private ChronicleMap<K, V> establishReplication(
VanillaChronicleMap<K, ?, ?, V, ?, ?> map, SimpleReplication simpleReplication,
ReplicationChannel channel) throws IOException {
private ChronicleMap<K, V> establishReplication(VanillaChronicleMap<K, ?, ?, V, ?, ?> 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<Replicator> 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<K, V,
VanillaChronicleMap.MAX_ENTRY_OVERSIZE_FACTOR);
}
int segmentHeaderSize() {
public byte identifier() {
return identifier;
}
public int segmentHeaderSize() {
int segments = actualSegments();
// reduce false sharing unless we have a lot of segments.
return segments <= 8192 ? 64 : 16;
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+4 -6
View File
@@ -17,8 +17,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.chronicle.hash.serialization.BytesWriter;
import net.openhft.chronicle.hash.serialization.AgileBytesMarshaller;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.serialization.BytesMarshallable;
import net.openhft.lang.io.serialization.BytesMarshaller;
@@ -224,10 +223,9 @@ public final class ChronicleMapBuilder<K, V>
return this;
}
public ChronicleMapBuilder<K, V> valueMarshallers(
BytesWriter<V> valueWriter, BytesReader<V> valueReader) {
valueBuilder.writer(valueWriter);
valueBuilder.reader(valueReader, valueBuilder.factory());
public ChronicleMapBuilder<K, V> valueMarshaller(
@NotNull AgileBytesMarshaller<V> valueMarshaller) {
valueBuilder.agileMarshaller(valueMarshaller, null);
return this;
}
}
src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+18 -19
View File
@@ -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<NodeDiscoveryEventListener> nodeDiscoveryEventListenerAtomicReference =
new AtomicReference<NodeDiscoveryEventListener>();
@@ -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<InetSocketAddress> toInetSocketCollection(Set<AddressAndPort> source)
throws UnknownHostException {
private InetSocketAddress[] toInetSocketArray(Set<AddressAndPort> source) throws
UnknownHostException {
// make a safe copy
final HashSet<AddressAndPort> addressAndPorts = new HashSet<AddressAndPort>(source);
if (addressAndPorts.isEmpty())
return Collections.emptyList();
return new InetSocketAddress[0];
ArrayList<InetSocketAddress> addresses =
new ArrayList<InetSocketAddress>(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 {
src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java
-2
View File
@@ -25,8 +25,6 @@ public final class OffHeapUpdatableChronicleMapBuilder<K, V>
public static <K, V> OffHeapUpdatableChronicleMapBuilder<K, V> of(
@NotNull Class<K> keyClass, @NotNull Class<V> valueClass) {
if (keyClass != CharSequence.class && keyClass.isInterface())
keyClass = DataValueClasses.directClassFor(keyClass);
if (!offHeapReference(valueClass)) {
if (!valueClass.isInterface()) {
throw new IllegalArgumentException(
src/main/java/net/openhft/chronicle/map/Replica.java
+1 -1
View File
@@ -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. <p> An identifier is used to
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+6 -5
View File
@@ -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 <K> the entries key type
* @param <V> the entries value type
*/
final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
V, VW, MVW extends MetaBytesWriter<V, VW>>
extends VanillaChronicleMap<K, KI, MKI, V, VW, MVW>
implements ChronicleMap<K, V>,
@@ -102,13 +102,12 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
private transient ModificationDelegator modificationDelegator;
private transient long startOfModificationIterators;
public ReplicatedChronicleMap(@NotNull AbstractChronicleMapBuilder<K, V, ?> builder,
byte identifier)
public ReplicatedChronicleMap(@NotNull AbstractChronicleMapBuilder<K, V, ?> 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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,
src/main/java/net/openhft/chronicle/map/Replicator.java
+1 -1
View File
@@ -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.
src/main/java/net/openhft/chronicle/map/Replicators.java
+7 -16
View File
@@ -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());
}
};
}
src/main/java/net/openhft/chronicle/map/SerializationBuilder.java
+16 -30
View File
@@ -83,29 +83,18 @@ final class SerializationBuilder<E> implements Cloneable {
NullObjectFactory.<E>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<E>) CharSequenceReader.of(), factory);
writer((BytesWriter<E>) CharSequenceWriter.instance());
} else if (eClass == Void.class) {
sizeMarshaller(VoidMarshaller.INSTANCE)
.reader((BytesReader<E>) VoidMarshaller.INSTANCE, factory)
.interop((BytesInterop<E>) VoidMarshaller.INSTANCE);
agileMarshaller((AgileBytesMarshaller<E>) VoidMarshaller.INSTANCE, factory);
} else if (eClass == Long.class) {
sizeMarshaller(LongMarshaller.INSTANCE)
.reader((BytesReader<E>) LongMarshaller.INSTANCE, factory)
.interop((BytesInterop<E>) LongMarshaller.INSTANCE);
agileMarshaller((AgileBytesMarshaller<E>) LongMarshaller.INSTANCE, factory);
} else if (eClass == Double.class) {
sizeMarshaller(DoubleMarshaller.INSTANCE)
.reader((BytesReader<E>) DoubleMarshaller.INSTANCE, factory)
.interop((BytesInterop<E>) DoubleMarshaller.INSTANCE);
agileMarshaller((AgileBytesMarshaller<E>) DoubleMarshaller.INSTANCE, factory);
} else if (eClass == Integer.class) {
sizeMarshaller(IntegerMarshaller.INSTANCE)
.reader((BytesReader<E>) IntegerMarshaller.INSTANCE, factory)
.interop((BytesInterop<E>) IntegerMarshaller.INSTANCE);
agileMarshaller((AgileBytesMarshaller<E>) IntegerMarshaller.INSTANCE, factory);
} else if (eClass == byte[].class) {
reader((BytesReader<E>) ByteArrayMarshaller.INSTANCE, factory);
interop((BytesInterop<E>) ByteArrayMarshaller.INSTANCE);
@@ -136,6 +125,14 @@ final class SerializationBuilder<E> implements Cloneable {
return this;
}
public SerializationBuilder<E> agileMarshaller(AgileBytesMarshaller<E> agileBytesMarshaller,
ObjectFactory<E> factory) {
if (factory == null) factory = this.factory;
return sizeMarshaller(agileBytesMarshaller)
.reader(agileBytesMarshaller, factory)
.interop(agileBytesMarshaller);
}
public SerializationBuilder<E> interop(BytesInterop<E> interop) {
return copyingInterop(null)
.setInterop(interop)
@@ -145,8 +142,6 @@ final class SerializationBuilder<E> implements Cloneable {
}
public SerializationBuilder<E> writer(BytesWriter<E> writer) {
if (writer instanceof BytesInterop)
return interop((BytesInterop<E>) writer);
return copyingInterop(CopyingInterop.FROM_WRITER)
.setInterop(writer)
.metaInterop(CopyingMetaBytesInterop
@@ -292,19 +287,10 @@ final class SerializationBuilder<E> 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<E> 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) {
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+7 -10
View File
@@ -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<K, V> implements ChronicleMap<K, V>, Closeable {
private volatile SocketChannel clientChannel;
private CloseablesManager closeables;
private final StatelessMapConfig builder;
private final StatelessBuilder builder;
private int maxEntrySize;
private final Class<K> kClass;
private final Class<V> vClass;
private final boolean putReturnsNull;
private final boolean removeReturnsNull;
static enum EventId {
HEARTBEAT,
STATEFUL_UPDATE,
@@ -90,20 +92,20 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V>(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<K, V> implements ChronicleMap<K, V>, 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");
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+22 -54
View File
@@ -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<K, V> {
switch (event) {
case LONG_SIZE:
return longSize(reader, writer, sizeLocation);
@@ -1300,10 +1292,6 @@ class StatelessServerConnector<K, V> {
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<K, V> {
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);
src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java
+2 -2
View File
@@ -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 {
src/main/java/net/openhft/chronicle/map/UdpReplicator.java
+3 -3
View File
@@ -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 {
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+20 -8
View File
@@ -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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
+ 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
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<K, KI, MKI extends MetaBytesInterop<K, KI>,
@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<K, KI, MKI extends MetaBytesInterop<K, KI>,
putWithoutLock(copies, metaKeyInterop, keyInterop, key(), value(),
segmentHash, true);
copies = null;
putOnClose = true;
Segment.this.writeUnlock();
}
src/main/java/net/openhft/chronicle/map/WriteContext.java
+2 -1
View File
@@ -21,9 +21,10 @@ package net.openhft.chronicle.map;
* @author Rob Austin.
*/
public interface WriteContext<K, V> extends Context<K, V> {
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();
}
src/main/java/net/openhft/chronicle/map/ZeroOutValueBytes.java
+1 -1
View File
@@ -29,7 +29,7 @@ final class ZeroOutValueBytes<K, V> implements PrepareValueBytes<K, V> {
@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);
}
}
src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
+25 -25
View File
@@ -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<E>
ChronicleSetBuilder(Class<E> keyClass) {
chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, DummyValue.class)
.valueMarshallers(DummyValueMarshaller.INSTANCE, DummyValueMarshaller.INSTANCE);
.valueMarshaller(DummyValueMarshaller.INSTANCE);
}
public static <K> ChronicleSetBuilder<K> of(Class<K> keyClass) {
@@ -170,6 +167,25 @@ public class ChronicleSetBuilder<E>
return this;
}
@Override
public ChronicleSetBuilder<E> replicators(byte identifier, ReplicationConfig... replicators) {
chronicleMapBuilder.replicators(identifier, replicators);
return this;
}
@Override
public ChronicleSetBuilder<E> channel(ChannelProvider.ChronicleChannel chronicleChannel) {
chronicleMapBuilder.channel(chronicleChannel);
return this;
}
@Override
public ChronicleSetBuilder<E> disableReplication() {
chronicleMapBuilder.disableReplication();
return this;
}
@Override
public String toString() {
return " ChronicleSetBuilder{" +
@@ -249,39 +265,23 @@ public class ChronicleSetBuilder<E>
}
@Override
public ChronicleSetBuilder<E> replication(SimpleReplication replication) {
chronicleMapBuilder.replication(replication);
public ChronicleSetBuilder<E> stateless(StatelessBuilder statelessBuilder) {
chronicleMapBuilder.stateless(statelessBuilder);
return this;
}
@Override
public ChronicleSetBuilder<E> replication(byte identifier, TcpConfig tcpTransportAndNetwork) {
chronicleMapBuilder.replication(identifier, tcpTransportAndNetwork);
public ChronicleSetBuilder<E> file(File file) throws IOException {
chronicleMapBuilder.file(file);
return this;
}
@Override
public StatelessClientConfig<ChronicleSet<E>> statelessClient(
InetSocketAddress remoteAddress) {
return new StatelessSetConfig<>(chronicleMapBuilder.statelessClient(remoteAddress));
}
@Override
public ChronicleHashInstanceConfig<ChronicleSet<E>> instance() {
return new InstanceConfig<>(chronicleMapBuilder.instance());
}
@Override
public ChronicleSet<E> create() throws IOException {
final ChronicleMap<E, DummyValue> map = chronicleMapBuilder.create();
return new SetFromMap<E>(map);
}
@Override
public ChronicleSet<E> createPersistedTo(File file) throws IOException {
ChronicleMap<E, DummyValue> map = chronicleMapBuilder.createPersistedTo(file);
return new SetFromMap<E>(map);
}
}
src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java
+3 -6
View File
@@ -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<DummyValue>, BytesReader<DummyValue>, SizeMarshaller {
enum DummyValueMarshaller implements AgileBytesMarshaller<DummyValue> {
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;
}