mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
Merge remote-tracking branch 'origin/master'
This commit is contained in:
@@ -56,7 +56,8 @@ public abstract class AbstractReplication {
|
||||
}
|
||||
|
||||
/**
|
||||
* Builder of {@link AbstractReplication} configurations.
|
||||
* Builder of {@link AbstractReplication} configurations. This class and it's subclasses are
|
||||
* mutable, configuration methods mutate the builder and return it back for convenient chaining.
|
||||
*
|
||||
* @param <R> the concrete {@link AbstractReplication} subclass: {@link
|
||||
* SingleChronicleHashReplication} or {@link ReplicationHub}
|
||||
@@ -77,6 +78,14 @@ public abstract class AbstractReplication {
|
||||
return (B) this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Configures UDP transport settings, used by Replications, created by this builder.
|
||||
* {@code null} means that UDP transport shouldn't being used.
|
||||
*
|
||||
* @param udpConfig the new UDP transport config for replications, created by this builder.
|
||||
* @return this builder back, for chaining
|
||||
* @see AbstractReplication#udpTransport()
|
||||
*/
|
||||
@NotNull
|
||||
public B udpTransport(@Nullable UdpTransportConfig udpConfig) {
|
||||
this.udpConfig = udpConfig;
|
||||
@@ -89,10 +98,24 @@ public abstract class AbstractReplication {
|
||||
return (B) this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a Replication instance with the given node (server) identifier.
|
||||
*
|
||||
* @param identifier the node (server) identifier of the returned replication
|
||||
* @return a new Replication instance with the specified node (server) identifier
|
||||
* @throws IllegalArgumentException if the given identifier is non-positive
|
||||
* @throws IllegalStateException if neither {@link #tcpTransportAndNetwork(
|
||||
* TcpTransportAndNetworkConfig)} nor {@link #udpTransport(UdpTransportConfig)} are
|
||||
* configured to non-{@code null}. At least one of the transport-level configs should be
|
||||
* specified.
|
||||
*/
|
||||
@NotNull
|
||||
public abstract R createWithId(byte identifier);
|
||||
|
||||
void check() {
|
||||
void check(byte identifier) {
|
||||
if (identifier <= 0)
|
||||
throw new IllegalArgumentException("Identifier must be positive, " + identifier +
|
||||
" given");
|
||||
if (udpConfig == null && tcpConfig == null)
|
||||
throw new IllegalStateException(
|
||||
"At least one transport method (TCP or UDP) should be configured");
|
||||
|
||||
@@ -17,6 +17,24 @@
|
||||
package net.openhft.chronicle.hash.replication;
|
||||
|
||||
|
||||
import net.openhft.chronicle.hash.ChronicleHash;
|
||||
import net.openhft.chronicle.hash.ChronicleHashBuilder;
|
||||
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
/**
|
||||
* A token which should be created via {@link ReplicationHub#createChannel(short)} call and passed
|
||||
* to {@link ChronicleHashInstanceConfig#replicatedViaChannel(ReplicationChannel)} method, to
|
||||
* establish a channel replication. See
|
||||
* <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
|
||||
* the corresponding section in ChronicleMap manual</a> for more information.
|
||||
*
|
||||
* <p>A {@code ReplicationChannel} could be used to replicate only one {@link ChronicleHash}
|
||||
* instance.
|
||||
*
|
||||
* @see ReplicationHub
|
||||
* @see ReplicationHub#createChannel(short)
|
||||
*/
|
||||
public final class ReplicationChannel {
|
||||
private ReplicationHub hub;
|
||||
private final short channelId;
|
||||
@@ -26,10 +44,25 @@ public final class ReplicationChannel {
|
||||
this.channelId = channelId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the {@link ReplicationHub} on which this {@code ReplicationChannel} was {@linkplain
|
||||
* ReplicationHub#createChannel(short) created}.
|
||||
*
|
||||
* @return the {@code ReplicationHub} to which this {@code ReplicationChannel} belongs
|
||||
* @see ReplicationHub#createChannel(short)
|
||||
*/
|
||||
@NotNull
|
||||
public ReplicationHub hub() {
|
||||
return hub;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the identifier of this channel, with which was {@linkplain
|
||||
* ReplicationHub#createChannel(short) created}.
|
||||
*
|
||||
* @return the identifier of this channel
|
||||
* @see ReplicationHub#createChannel(short)
|
||||
*/
|
||||
public short channelId() {
|
||||
return channelId;
|
||||
}
|
||||
|
||||
@@ -16,10 +16,46 @@
|
||||
|
||||
package net.openhft.chronicle.hash.replication;
|
||||
|
||||
import net.openhft.chronicle.hash.ChronicleHash;
|
||||
import net.openhft.chronicle.hash.ChronicleHashBuilder;
|
||||
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
|
||||
import net.openhft.chronicle.map.ChronicleMap;
|
||||
import net.openhft.chronicle.set.ChronicleSet;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
/**
|
||||
* This class holds all configurations of
|
||||
* <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
|
||||
* multicontainer replication</a>, which is usable, when you want to replicate several {@linkplain
|
||||
* ChronicleHash ChronicleHashes} ({@linkplain ChronicleMap maps}, {@linkplain ChronicleSet sets},
|
||||
* etc.) between same servers. {@code ReplicationHub} allows to share TCP/UDP connection, sockets,
|
||||
* buffers, worker threads for that, considerably reducing resources usage and increasing cumulative
|
||||
* (for all replicated containers) replication throughput. See
|
||||
* <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
|
||||
* the corresponding section in ChronicleMap manual</a> for more information.
|
||||
*
|
||||
* <p>Create instances of this class using this pattern: <pre>{@code
|
||||
* ReplicationHub hub = ReplicationHub.builder()
|
||||
* .tcpTransportAndNetwork(tcpConfig)
|
||||
* // more configurations...
|
||||
* .createWithId(identifierOfThisServerWithinTheChronicleReplicationNetwork);}</pre>
|
||||
* Then, given you prepared a {@link ChronicleHashBuilder builder} to create a {@link ChronicleHash
|
||||
* }, configure it's replication via channel of this hub like this: <pre>{@code
|
||||
* ChronicleMap myMap = builder.instance()
|
||||
* .replicatedViaChannel(hub.createChannel(myMapChannelId))
|
||||
* .persistedTo(myMapFile) // optional, for this example
|
||||
* .create();}</pre>
|
||||
*
|
||||
* @see ChronicleHashInstanceConfig#replicatedViaChannel(ReplicationChannel)
|
||||
* @see ReplicationChannel
|
||||
*/
|
||||
public final class ReplicationHub extends AbstractReplication {
|
||||
|
||||
/**
|
||||
* Creates and returns a new {@link ReplicationHub.Builder}.
|
||||
*
|
||||
* @return a new {@link ReplicationHub.Builder}
|
||||
*/
|
||||
@NotNull
|
||||
public static Builder builder() {
|
||||
return new Builder();
|
||||
@@ -34,14 +70,48 @@ public final class ReplicationHub extends AbstractReplication {
|
||||
channels = new ReplicationChannel[builder.maxNumberOfChronicles];
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of {@link
|
||||
* ChronicleHash}es, replicated by this {@code ReplicationHub}, could have. {@code
|
||||
* RuntimeException} is thrown on attempt to replicate with this {@code ReplicationHub}
|
||||
* a {@code ChronicleHash} with larger entry size.
|
||||
*
|
||||
* @return the maximum entry size
|
||||
* @see Builder#maxEntrySize(int)
|
||||
*/
|
||||
public int maxEntrySize() {
|
||||
return maxEntrySize;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the maximum number of channels could be {@linkplain #createChannel(short) created}
|
||||
* for this {@code ReplicationHub}.
|
||||
*
|
||||
* @return the maximum number of channels
|
||||
* @see Builder#maxNumberOfChannels()
|
||||
* @see #createChannel(short)
|
||||
*/
|
||||
public int maxNumberOfChannels() {
|
||||
return channels.length;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new {@link ReplicationChannel} in this {@code ReplicationHub} with the given
|
||||
* identifier. Identifier shouldn't be lesser than zero and greater or equal to
|
||||
* {@code maxNumberOfChannels() - 1}. On a {@code ReplicationHub} instance,
|
||||
* {@code ReplicationChannel} could be created only once for each possible {@code channelId}
|
||||
* value.
|
||||
*
|
||||
* @param channelId the identifier of the channel. Should be equal for replicated containers
|
||||
* on different nodes (servers)
|
||||
* @return a new {@code ReplicationChannel} instance, in fact just incapsulating the given
|
||||
* {@code channelId} for this {@code ReplicationHub}
|
||||
* @throws IllegalArgumentException if the specified {@code channelId} is out of<br>
|
||||
* <code>[0, {@link #maxNumberOfChannels()})</code> range
|
||||
* @throws IllegalStateException if {@code ReplicationChannel} with the specified
|
||||
* {@code channelId} has already been acquired on this {@code ReplicationHub}
|
||||
* @see ReplicationChannel
|
||||
*/
|
||||
public synchronized ReplicationChannel createChannel(short channelId) {
|
||||
if (channelId < 0)
|
||||
throw new IllegalArgumentException("channelId should be positive");
|
||||
@@ -49,25 +119,50 @@ public final class ReplicationHub extends AbstractReplication {
|
||||
throw new IllegalArgumentException("maxNumberOfChannels is configured (or defaulted) " +
|
||||
"to " + maxNumberOfChannels() + ", channelId=" + channelId + " is requested");
|
||||
if (channels[channelId] != null)
|
||||
throw new IllegalArgumentException("The requested channelId=" + channelId +
|
||||
throw new IllegalStateException("The requested channelId=" + channelId +
|
||||
" is already in use");
|
||||
ReplicationChannel channel = new ReplicationChannel(this, channelId);
|
||||
channels[channelId] = channel;
|
||||
return channel;
|
||||
}
|
||||
|
||||
/**
|
||||
* Builder of {@link ReplicationHub}s.
|
||||
*/
|
||||
public static final class Builder extends AbstractReplication.Builder<ReplicationHub, Builder> {
|
||||
private int maxEntrySize = 1024;
|
||||
private int maxNumberOfChronicles = 128;
|
||||
|
||||
private Builder() {}
|
||||
|
||||
/**
|
||||
* Configures the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of
|
||||
* {@link ChronicleHash}es, replicated via {@code ReplicationHub}s, created by this builder.
|
||||
*
|
||||
* <p>Default value is {@code 1024}.
|
||||
*
|
||||
* @param maxEntrySize {@link ReplicationHub#maxEntrySize()} of {@link ReplicationHub}s,
|
||||
* created by this builder
|
||||
* @return this builder object back, for chaining
|
||||
* @see ReplicationHub#maxEntrySize()
|
||||
*/
|
||||
@NotNull
|
||||
public Builder maxEntrySize(int maxEntrySize) {
|
||||
this.maxEntrySize = maxEntrySize;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Configures the maximum number of channels could be {@linkplain #createChannel(short)
|
||||
* created for {@code ReplicationHub}s, created by this builder.
|
||||
*
|
||||
* <p>Default value is {@code 128}.
|
||||
*
|
||||
* @param maxNumberOfChannels {@link ReplicationHub#maxNumberOfChannels()} of
|
||||
* {@link ReplicationHub}s, created by this builder
|
||||
* @return this builder object back, for chaining
|
||||
* @see ReplicationHub#maxNumberOfChannels()
|
||||
*/
|
||||
@NotNull
|
||||
public Builder maxNumberOfChannels(int maxNumberOfChannels) {
|
||||
this.maxNumberOfChronicles = maxNumberOfChannels;
|
||||
@@ -77,7 +172,7 @@ public final class ReplicationHub extends AbstractReplication {
|
||||
@NotNull
|
||||
@Override
|
||||
public ReplicationHub createWithId(byte identifier) {
|
||||
check();
|
||||
check(identifier);
|
||||
return new ReplicationHub(identifier, this);
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -36,7 +36,7 @@ public final class SingleChronicleHashReplication extends AbstractReplication {
|
||||
@NotNull
|
||||
@Override
|
||||
public SingleChronicleHashReplication createWithId(byte identifier) {
|
||||
check();
|
||||
check(identifier);
|
||||
return new SingleChronicleHashReplication(identifier, this);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -501,8 +501,6 @@ final class ChannelProvider implements Closeable {
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
// synchronized on parent instance to prevent possible (but very unlikely) contention
|
||||
// between close() calls on different channels
|
||||
channelDataLock.writeLock().lock();
|
||||
try {
|
||||
chronicleChannelList.remove(chronicleChannels[chronicleChannel]);
|
||||
|
||||
@@ -32,6 +32,8 @@ final class InstanceConfig<K, V>
|
||||
final ReplicationChannel channel;
|
||||
final File file;
|
||||
|
||||
private boolean used = false;
|
||||
|
||||
InstanceConfig(AbstractChronicleMapBuilder<K, V, ?> mapBuilder,
|
||||
SingleChronicleHashReplication singleHashReplication,
|
||||
ReplicationChannel channel,
|
||||
@@ -67,7 +69,15 @@ final class InstanceConfig<K, V>
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChronicleMap<K, V> create() throws IOException {
|
||||
public synchronized ChronicleMap<K, V> create() throws IOException {
|
||||
if (used)
|
||||
throw new IllegalStateException(
|
||||
"A ChronicleMap has already been created using this instance config. " +
|
||||
"Create a new instance config (builder.instance() or call any " +
|
||||
"configuration method on this instance config) to create a new " +
|
||||
"ChronicleMap instance");
|
||||
used = true;
|
||||
return mapBuilder.create(this);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -29,6 +29,8 @@ final class InstanceConfig<E> implements ChronicleHashInstanceConfig<ChronicleSe
|
||||
|
||||
private final ChronicleHashInstanceConfig<ChronicleMap<E, DummyValue>> mapConfig;
|
||||
|
||||
private boolean used;
|
||||
|
||||
InstanceConfig(ChronicleHashInstanceConfig<ChronicleMap<E, DummyValue>> mapConfig) {
|
||||
this.mapConfig = mapConfig;
|
||||
}
|
||||
@@ -56,7 +58,14 @@ final class InstanceConfig<E> implements ChronicleHashInstanceConfig<ChronicleSe
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChronicleSet<E> create() throws IOException {
|
||||
public synchronized ChronicleSet<E> create() throws IOException {
|
||||
if (used)
|
||||
throw new IllegalStateException(
|
||||
"A ChronicleSet has already been created using this instance config. " +
|
||||
"Create a new instance config (builder.instance() or call any " +
|
||||
"configuration method on this instance config) to create a new " +
|
||||
"ChronicleSet instance");
|
||||
used = true;
|
||||
return new SetFromMap<>(mapConfig.create());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user