HCOLL-134 checking in a alpha version of this code that works with the replication hub

This commit is contained in:
Rob Austin
2014-11-18 11:36:26 +00:00
parent f0c099cd62
commit a210c35263
14 changed files with 169 additions and 115 deletions
@@ -0,0 +1,19 @@
package net.openhft.chronicle.hash;
import java.io.IOException;
/**
* @author Rob Austin.
*/
public interface FindByName {
/**
* @param name the name of the map or set
* @param <T>
* @return a chronicle map or set
* @throws IllegalArgumentException if a map with this name can not be found
* @throws IOException if it not possible to create the map or set
*/
<T extends ChronicleHash> T create(String name) throws IllegalArgumentException,
IOException;
}
@@ -19,10 +19,12 @@ package net.openhft.chronicle.hash.replication;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import java.io.Serializable;
/**
* Common configurations of {@link SingleChronicleHashReplication} and {@link ReplicationHub}.
*/
public abstract class AbstractReplication {
public abstract class AbstractReplication implements Serializable{
private final byte localIdentifier;
private final @Nullable TcpTransportAndNetworkConfig tcpConfig;
private final @Nullable UdpTransportConfig udpConfig;
@@ -18,6 +18,7 @@
package net.openhft.chronicle.hash.replication;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.TimeUnit;
@@ -26,7 +27,7 @@ import static java.util.Collections.unmodifiableSet;
import static java.util.concurrent.TimeUnit.SECONDS;
public final class TcpTransportAndNetworkConfig {
public final class TcpTransportAndNetworkConfig implements Serializable {
private static final int DEFAULT_PACKET_SIZE = 1024 * 8;
private static final long DEFAULT_HEART_BEAT_INTERVAL = 20;
@@ -19,11 +19,12 @@
package net.openhft.chronicle.hash.replication;
import java.io.Serializable;
import java.util.concurrent.TimeUnit;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
public final class ThrottlingConfig {
public final class ThrottlingConfig implements Serializable {
private static final long DEFAULT_BUCKET_INTERVAL = 100L;
private static final TimeUnit DEFAULT_BUCKET_INTERVAL_UNIT = MILLISECONDS;
@@ -49,6 +49,14 @@ public final class UdpTransportConfig {
this.throttlingConfig = throttlingConfig;
}
public static UdpTransportConfig multiCast(@NotNull InetAddress address, int port,
@NotNull NetworkInterface networkInterface) {
if (!address.isMulticastAddress() || networkInterface == null)
throw new IllegalArgumentException();
return create(address, port, networkInterface, ThrottlingConfig.noThrottling());
}
public static UdpTransportConfig of(@NotNull InetAddress address, int port) {
if (address.isMulticastAddress())
throw new IllegalArgumentException();
@@ -534,8 +534,9 @@ abstract class AbstractChronicleMapBuilder<K, V,
@Override
public String toString() {
return "ChronicleMapOnHeapUpdatableBuilder{" +
"actualSegments=" + pretty(actualSegments) +
return "ChronicleMapBuilder{" +
"name=" + name +
", actualSegments=" + pretty(actualSegments) +
", minSegments=" + pretty(minSegments) +
", actualEntriesPerSegment=" + pretty(actualEntriesPerSegment) +
", keySize=" + pretty(keySize) +
@@ -961,7 +962,6 @@ abstract class AbstractChronicleMapBuilder<K, V,
return this;
}
public String name() {
return this.name;
}
@@ -16,6 +16,8 @@ import net.openhft.lang.io.serialization.ObjectSerializer;
import net.openhft.lang.model.Byteable;
import net.openhft.lang.model.DataValueClasses;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
@@ -53,18 +55,17 @@ public final class ChronicleMapBuilder<K, V> implements ChronicleMapBuilderI<K,
if (valueClass.isEnum())
return new ChronicleMapBuilder<K, V>(OnHeapUpdatableChronicleMapBuilder.of(keyClass, valueClass));
if (keyClass.isInterface() && !builtInType(keyClass)) {
if (keyClass.isInterface() && !builtInType(keyClass))
keyClass = DataValueClasses.directClassFor(keyClass);
}
if ((valueClass.isInterface() && !builtInType(valueClass))) {
valueClass = DataValueClasses.directClassFor(valueClass);
} else if (!offHeapReference(valueClass)) {
return new ChronicleMapBuilder<K, V>(OnHeapUpdatableChronicleMapBuilder.of(keyClass, valueClass));
}
OffHeapUpdatableChronicleMapBuilder<K, V> kvOffHeapUpdatableChronicleMapBuilder = new
OffHeapUpdatableChronicleMapBuilder<K, V>(keyClass, valueClass);
return new ChronicleMapBuilder<K, V>(kvOffHeapUpdatableChronicleMapBuilder);
ChronicleMapBuilderI<K, V> builder = new OffHeapUpdatableChronicleMapBuilder<K, V>(keyClass, valueClass);
return new ChronicleMapBuilder<K, V>(builder);
}
static boolean builtInType(Class clazz) {
@@ -337,6 +338,7 @@ public final class ChronicleMapBuilder<K, V> implements ChronicleMapBuilderI<K,
/**
* each map can be identified by a name
*
* @param name the name of the map
* @return self
*/
@@ -346,9 +348,10 @@ public final class ChronicleMapBuilder<K, V> implements ChronicleMapBuilderI<K,
return this;
}
@Override
public String name() {
return delegate.name() ;
return delegate.name();
}
@Override
@@ -361,6 +364,18 @@ public final class ChronicleMapBuilder<K, V> implements ChronicleMapBuilderI<K,
@Override
public ChronicleMap<K, V> create() {
return delegate.create();
}
private void registerBuilderWithReplicationHub() {
try {
FindFindByName mapByName = new NodeDiscovery().mapByName();
mapByName.add(this);
} catch (Exception e) {
LOG.error("", e);
}
}
@Override
@@ -488,14 +503,9 @@ public final class ChronicleMapBuilder<K, V> implements ChronicleMapBuilderI<K,
return this;
}
@Override
public ChronicleMap<K, V> createReplicated(byte b) throws IOException {
return delegate.createReplicated(b);
}
@Override
public ChronicleMap<K, V> createPersistedTo(File file) throws IOException {
return delegate.createPersistedTo(file);
return delegate.createPersistedTo(file);
}
}
@@ -200,7 +200,7 @@ interface ChronicleMapBuilderI<K, V> extends Serializable {
* @see ChronicleMapBuilder
* @see #objectSerializer(ObjectSerializer)
*/
ChronicleMapBuilderI<K, V> valueMarshaller(
ChronicleMapBuilderI<K, V> valueMarshaller(
@NotNull BytesMarshaller<V> valueMarshaller);
@@ -292,12 +292,12 @@ interface ChronicleMapBuilderI<K, V> extends Serializable {
ChronicleMapBuilderI<K, V> putReturnsNull(boolean b);
ChronicleMap<K, V> createReplicated(byte b) throws IOException;
ChronicleMapBuilderI<K, V> removeReturnsNull(boolean b);
ChronicleMapBuilderI<K, V> name(String name);
ChronicleMapBuilderI<K, V> name(String name);
String name();
@@ -18,6 +18,7 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import net.openhft.chronicle.hash.FindByName;
import net.openhft.chronicle.hash.replication.ReplicationHub;
import java.io.File;
@@ -30,7 +31,7 @@ import static net.openhft.chronicle.map.ChronicleMapBuilder.of;
/**
* @author Rob Austin.
*/
class FindMapByName {
class FindFindByName implements FindByName {
public static final int MAP_BY_NAME_CHANNEL = 1;
private final ReplicationHub replicationHub;
@@ -42,7 +43,7 @@ class FindMapByName {
/**
* @throws IOException
*/
public FindMapByName(ReplicationHub replicationHub) throws IOException {
public FindFindByName(ReplicationHub replicationHub) throws IOException {
this.replicationHub = replicationHub;
this.map = (Map) of(CharSequence.class, ChronicleMapBuilder.class)
.entrySize(300)
@@ -54,7 +55,7 @@ class FindMapByName {
ChronicleMapBuilder chronicleMapBuilder = map.get(name);
if (chronicleMapBuilder == null)
throw new IllegalArgumentException("A map of this name can not be found name=" + name);
throw new IllegalArgumentException("A map name=" + name + " can not be found.");
return chronicleMapBuilder;
}
@@ -43,7 +43,6 @@ 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.UdpTransportConfig.of;
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,22 +55,23 @@ import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG;
*/
public class NodeDiscovery {
public static final int DEFAULT_UDP_BROADCAST_PORT = 8124;
public static final int DEFAULT_UDP_BROADCAST_PORT = 8129;
public static final short DEFAULT_TCP_PORT = (short) 8123;
public static final int SERIALIZED_ENTRY_SIZE = 1024;
private final AddressAndPort ourAddressAndPort;
private final UdpTransportConfig udpConfig;
public static final InetAddress DEFAULT_UDP_MASK;
public static final InetAddress IP_MULIT_CAST_GROUP;
static {
InetAddress mask = null;
try {
mask = Inet4Address.getByName("255.255.255.255");
mask = Inet4Address.getByName("224.0.8.0");
} catch (Exception e) {
mask = null;
}
DEFAULT_UDP_MASK = mask;
IP_MULIT_CAST_GROUP = mask;
}
@@ -79,18 +79,21 @@ public class NodeDiscovery {
private final DiscoveryNodeBytesMarshallable discoveryNodeBytesMarshallable;
private final AtomicReference<NodeDiscoveryEventListener> nodeDiscoveryEventListenerAtomicReference =
new AtomicReference<NodeDiscoveryEventListener>();
private KnownNodes knownNodes;
public NodeDiscovery() throws IOException {
this((short) DEFAULT_UDP_BROADCAST_PORT, DEFAULT_TCP_PORT, getDefaultAddress(), DEFAULT_UDP_MASK);
this((short) DEFAULT_UDP_BROADCAST_PORT, DEFAULT_TCP_PORT, getDefaultAddress(), IP_MULIT_CAST_GROUP);
}
public NodeDiscovery(short udpBroadcastPort,
short tcpPort,
@NotNull InetAddress tcpAddress,
@NotNull InetAddress udpBroadcastAddress) throws IOException {
@NotNull InetAddress udpMultiCastGroup) throws IOException {
this.ourAddressAndPort = new AddressAndPort(tcpAddress.getAddress(), tcpPort);
this.udpConfig = of(udpBroadcastAddress, udpBroadcastPort);
this.udpConfig = UdpTransportConfig.multiCast(udpMultiCastGroup, udpBroadcastPort,
ConcurrentExpiryMap.defaultNetworkInterface());
knownNodes = new KnownNodes();
discoveryNodeBytesMarshallable = new DiscoveryNodeBytesMarshallable
@@ -103,7 +106,7 @@ public class NodeDiscovery {
}
public synchronized FindMapByName discoverMap() throws
public synchronized FindFindByName mapByName() throws
IOException, InterruptedException {
final AtomicInteger ourProposedIdentifier = new AtomicInteger();
@@ -252,7 +255,7 @@ public class NodeDiscovery {
.tcpTransportAndNetwork(tcpConfig)
.createWithId(identifier);
return new FindMapByName(replicationHub);
return new FindFindByName(replicationHub);
}
@@ -25,7 +25,7 @@ import net.openhft.lang.model.DataValueClasses;
import org.jetbrains.annotations.NotNull;
final class OffHeapUpdatableChronicleMapBuilder<K, V>
extends AbstractChronicleMapBuilder<K, V, OffHeapUpdatableChronicleMapBuilder<K, V>> implements ChronicleMapBuilderI<K, V> {
extends AbstractChronicleMapBuilder<K, V, OffHeapUpdatableChronicleMapBuilder<K, V>> {
public static <K, V> ChronicleMapBuilderI<K, V> of(
@NotNull Class<K> keyClass, @NotNull Class<V> valueClass) {
@@ -37,10 +37,9 @@ import java.io.Serializable;
* non-{@link Byteable} values, which don't point to off-heap memory directly, including primitives ({@link
* Long}, {@link Double}, etc.), {@link String}s and {@link CharSequence}s, values implementing {@link
* BytesMarshallable}, {@link Externalizable} or {@link Serializable} interface, or any other values for which
* {@linkplain #valueMarshaller(BytesMarshaller) custom marshaller} is provided.
* <p/>
* <p>Use static {@link #of(Class, Class) of(Key.class, Value.class)} method to obtain a {@code
* ChronicleMapOnHeapUpdatableBuilder} instance.
* {@linkplain #valueMarshaller(BytesMarshaller) custom marshaller} is provided. <p/> <p>Use static {@link
* #of(Class, Class) of(Key.class, Value.class)} method to obtain a {@code ChronicleMapOnHeapUpdatableBuilder}
* instance.
*
* @param <K> key type of the maps, created by this builder
* @param <V> value type of the maps, created by this builder
@@ -49,15 +48,13 @@ import java.io.Serializable;
* @see ChronicleHashBuilder
*/
final class OnHeapUpdatableChronicleMapBuilder<K, V>
extends AbstractChronicleMapBuilder<K, V, OnHeapUpdatableChronicleMapBuilder<K, V>> implements
ChronicleMapBuilderI<K, V> {
extends AbstractChronicleMapBuilder<K, V, OnHeapUpdatableChronicleMapBuilder<K, V>> {
/**
* Returns a new {@code ChronicleMapOnHeapUpdatableBuilder} instance which is able to {@linkplain
* #create() create} maps with the specified key and value classes.
* <p/>
* <p>{@code ChronicleMapOnHeapUpdatableBuilder} analyzes provided key and value classes and automatically
* chooses the most specific and effective serializer which it is aware about. Read <a
* #create() create} maps with the specified key and value classes. <p/> <p>{@code
* ChronicleMapOnHeapUpdatableBuilder} analyzes provided key and value classes and automatically chooses
* the most specific and effective serializer which it is aware about. Read <a
* href="https://github.com/OpenHFT/Chronicle-Map#serialization">the section about serialization in
* Chronicle Map manual</a> for more information.
*
@@ -107,14 +104,11 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
/**
* Configures the optimal number of bytes, taken by serialized form of values, put into maps, created by
* this builder. If value size is always the same, call {@link #constantValueSizeBySample(Object)} method
* instead of this one.
* <p/>
* <p>If value is a boxed primitive type, i. e. if value size is known statically, it is automatically
* accounted and shouldn't be specified by user.
* <p/>
* <p>If value size varies moderately, specify the size higher than average, but lower than the maximum
* possible, to minimize average memory overuse. If value size varies in a wide range, it's better to use
* {@linkplain #entrySize(int) entry size} in "chunk" mode and configure it directly.
* instead of this one. <p/> <p>If value is a boxed primitive type, i. e. if value size is known
* statically, it is automatically accounted and shouldn't be specified by user. <p/> <p>If value size
* varies moderately, specify the size higher than average, but lower than the maximum possible, to
* minimize average memory overuse. If value size varies in a wide range, it's better to use {@linkplain
* #entrySize(int) entry size} in "chunk" mode and configure it directly.
*
* @param valueSize number of bytes, taken by serialized form of values
* @return this {@code ChronicleMapOnHeapUpdatableBuilder} back
@@ -130,13 +124,10 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
/**
* Configures the constant number of bytes, taken by serialized form of values, put into maps, created by
* this builder. This is done by providing the {@code sampleValue}, all values should take the same number
* of bytes in serialized form, as this sample object.
* <p/>
* <p>If values are of boxed primitive type or {@link Byteable} subclass, i. e. if value size is known
* statically, it is automatically accounted and this method shouldn't be called.
* <p/>
* <p>If value size varies, method {@link #valueSize(int)} or {@link #entrySize(int)} should be called
* instead of this one.
* of bytes in serialized form, as this sample object. <p/> <p>If values are of boxed primitive type or
* {@link Byteable} subclass, i. e. if value size is known statically, it is automatically accounted and
* this method shouldn't be called. <p/> <p>If value size varies, method {@link #valueSize(int)} or {@link
* #entrySize(int)} should be called instead of this one.
*
* @param sampleValue the sample value
* @return this builder back
@@ -151,8 +142,7 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
}
/**
* {@inheritDoc}
* <p/>
* {@inheritDoc} <p/>
* <p>Example: <pre>{@code Map<Key, Value> map =
* ChronicleMapBuilder.of(Key.class, Value.class)
* .entries(1_000_000)
@@ -160,9 +150,8 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
* // this class hasn't implemented yet, just for example
* .objectSerializer(new KryoObjectSerializer())
* .create();}</pre>
* <p/>
* <p>This serializer is used to serialize both keys and values, if they both require this: loosely typed,
* nullable, and custom {@linkplain #keyMarshaller(BytesMarshaller) key} and {@linkplain
* <p/> <p>This serializer is used to serialize both keys and values, if they both require this: loosely
* typed, nullable, and custom {@linkplain #keyMarshaller(BytesMarshaller) key} and {@linkplain
* #valueMarshaller(BytesMarshaller) value} marshallers are not configured.
*/
@Override
@@ -172,14 +161,12 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
}
/**
* {@inheritDoc}
* <p/>
* <p>If {@linkplain #valueMarshaller(BytesMarshaller) custom value marshaller} is configured, this
* configuration is unused, because it is incapsulated in {@link BytesMarshaller#read(Bytes)} method
* (without provided instance to read the data into), i. e. it's is the user-side responsibility. Actually
* this is just a convenience method supporting value marshaller configurations, made initially during
* {@link #of(Class, Class)} call, if the value class is {@link BytesMarshallable} or {@link
* Externalizable} subclass.
* {@inheritDoc} <p/> <p>If {@linkplain #valueMarshaller(BytesMarshaller) custom value marshaller} is
* configured, this configuration is unused, because it is incapsulated in {@link
* BytesMarshaller#read(Bytes)} method (without provided instance to read the data into), i. e. it's is
* the user-side responsibility. Actually this is just a convenience method supporting value marshaller
* configurations, made initially during {@link #of(Class, Class)} call, if the value class is {@link
* BytesMarshallable} or {@link Externalizable} subclass.
*
* @throws IllegalStateException if custom value marshaller is specified or value class is not either
* {@code BytesMarshallable} or {@code Externalizable}
@@ -191,9 +178,7 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
}
/**
* {@inheritDoc}
* <p/>
* <p>By default, the default value is set to {@code null}.
* {@inheritDoc} <p/> <p>By default, the default value is set to {@code null}.
*
* @see #defaultValueProvider(DefaultValueProvider)
*/
@@ -203,10 +188,8 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
}
/**
* {@inheritDoc}
* <p/>
* <p>By default, default value provider is not specified, {@link #defaultValue(Object) default value} is
* specified instead.
* {@inheritDoc} <p/> <p>By default, default value provider is not specified, {@link #defaultValue(Object)
* default value} is specified instead.
*
* @see #defaultValue(Object)
*/
@@ -236,13 +219,11 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
/**
* Configures the marshallers, used to serialize/deserialize values to/from off-heap memory in maps,
* created by this builder. See <a href="https://github.com/OpenHFT/Chronicle-Map#serialization">the
* section about serialization in ChronicleMap manual</a> for more information.
* <p/>
* <p>Configuring marshalling this way results to a little bit more compact in-memory layout of the map,
* comparing to a single interface configuration: {@link #valueMarshaller(BytesMarshaller)}.
* <p/>
* <p>Passing {@link net.openhft.chronicle.hash.serialization.BytesInterop} instead of plain {@link
* BytesWriter} is, of cause, possible, but currently pointless for values.
* section about serialization in ChronicleMap manual</a> for more information. <p/> <p>Configuring
* marshalling this way results to a little bit more compact in-memory layout of the map, comparing to a
* single interface configuration: {@link #valueMarshaller(BytesMarshaller)}. <p/> <p>Passing {@link
* net.openhft.chronicle.hash.serialization.BytesInterop} instead of plain {@link BytesWriter} is, of
* cause, possible, but currently pointless for values.
*
* @param valueWriter the new value object &rarr; {@link Bytes} writer (interop) strategy
* @param valueReader the new {@link Bytes} &rarr; value object reader strategy
@@ -258,15 +239,14 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
/**
* Configures the marshaller used to serialize actual value sizes to off-heap memory in maps, created by
* this builder.
* <p/>
* <p>Default value size marshaller is so-called {@linkplain net.openhft.chronicle.hash.serialization.SizeMarshallers#stopBit()
* stop bit encoding marshalling}. If {@linkplain #constantValueSizeBySample(Object) constant value size}
* is configured, or defaulted if the value type is always constant and {@code ChronicleHashBuilder}
* implementation knows about it, this configuration takes no effect, because a special {@link
* SizeMarshaller} implementation, which doesn't actually do any marshalling, and just returns the known
* constant size on {@link SizeMarshaller#readSize(Bytes)} calls, is used instead of any {@code
* SizeMarshaller} configured using this method.
* this builder. <p/> <p>Default value size marshaller is so-called {@linkplain
* net.openhft.chronicle.hash.serialization.SizeMarshallers#stopBit() stop bit encoding marshalling}. If
* {@linkplain #constantValueSizeBySample(Object) constant value size} is configured, or defaulted if the
* value type is always constant and {@code ChronicleHashBuilder} implementation knows about it, this
* configuration takes no effect, because a special {@link SizeMarshaller} implementation, which doesn't
* actually do any marshalling, and just returns the known constant size on {@link
* SizeMarshaller#readSize(Bytes)} calls, is used instead of any {@code SizeMarshaller} configured using
* this method.
*
* @param valueSizeMarshaller the new marshaller, used to serialize actual value sizes to off-heap memory
* @return this builder back
@@ -280,9 +260,8 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
/**
* Configures the procedure which is called on the bytes, which later the returned value is pointing to,
* if the key is absent, on {@link ChronicleMap#acquireUsing(Object, Object) acquireUsing()} call on maps,
* created by this builder. See {@link PrepareValueBytes} for more information.
* <p/>
* <p>The default preparation callback zeroes out the value bytes.
* created by this builder. See {@link PrepareValueBytes} for more information. <p/> <p>The default
* preparation callback zeroes out the value bytes.
*
* @param prepareValueBytes what to do with the value bytes before assigning them into the {@link
* Byteable} value to return from {@code acquireUsing()} call