From 91a513b1115b9b2a04a76af6a59e642470ed7eb3 Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 29 Oct 2014 18:44:56 +0300 Subject: [PATCH] HCOLL-169 reduced replication API (not finished) --- .../chronicle/hash/ChannelProvider.java | 21 ++++++------ .../chronicle/hash/TcpReplicationConfig.java | 9 ++--- .../chronicle/hash/ThrottlingConfig.java | 33 ++++--------------- .../map/AbstractChannelReplicator.java | 2 +- .../map/AbstractChronicleMapBuilder.java | 13 +++++++- .../openhft/chronicle/map/TcpReplicator.java | 7 ++-- 6 files changed, 36 insertions(+), 49 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/ChannelProvider.java b/src/main/java/net/openhft/chronicle/hash/ChannelProvider.java index 3f4334939..54a621cf2 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/hash/ChannelProvider.java @@ -169,8 +169,7 @@ public final class ChannelProvider implements Closeable { newBitSet(systemModificationIterator.length()); private final AtomicReferenceArray modificationIterator = new AtomicReferenceArray(128); - private final Set replicators = - new CopyOnWriteArraySet(); + private final Set replicators = new CopyOnWriteArraySet(); ChannelProvider(ChannelProviderBuilder builder) { localIdentifier = builder.identifier; @@ -195,6 +194,10 @@ public final class ChannelProvider implements Closeable { add((short) 0, systemMessageQueue.asReplica, systemMessageQueue.asEntryExternalizable); } + public int maxEntrySize() { + return maxEntrySize; + } + /** * creates a bit set based on a number of bits * @@ -271,7 +274,7 @@ public final class ChannelProvider implements Closeable { @Override public void close() throws IOException { - for (AbstractChannelReplicator replicator : replicators) { + for (Closeable replicator : replicators) { replicator.close(); } // indexed reverse loop to avoid ConcurrentModificationException and remove elements from @@ -282,7 +285,7 @@ public final class ChannelProvider implements Closeable { } } - void add(AbstractChannelReplicator replicator) { + void add(Closeable replicator) { replicators.add(replicator); } @@ -413,16 +416,14 @@ public final class ChannelProvider implements Closeable { protected Closeable applyTo(AbstractChronicleMapBuilder builder, Replica map, EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap) { - if (builder.entrySize() > maxEntrySize) { - throw new IllegalArgumentException("During ReplicatingChannelBuilder setup, " + - "maxEntrySize=" + maxEntrySize + " was specified, but map with " + - "entrySize=" + builder.entrySize() + " is attempted to apply" + - "to the replicator"); - } add(chronicleChannel, map, entryExternalizable); return this; } + public ChannelProvider provider() { + return ChannelProvider.this; + } + @Override public void close() throws IOException { chronicleChannelList.remove(chronicleChannels[chronicleChannel]); diff --git a/src/main/java/net/openhft/chronicle/hash/TcpReplicationConfig.java b/src/main/java/net/openhft/chronicle/hash/TcpReplicationConfig.java index c187dec5b..22e4aa6de 100644 --- a/src/main/java/net/openhft/chronicle/hash/TcpReplicationConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/TcpReplicationConfig.java @@ -80,13 +80,8 @@ public class TcpReplicationConfig implements ReplicationConfig { return this; } - - public long heartBeatInterval() { - return heartBeatInterval; - } - - public TimeUnit heartBeatIntervalUnit() { - return heartBeatIntervalUnit; + public long heartBeatInterval(TimeUnit unit) { + return unit.convert(heartBeatInterval, heartBeatIntervalUnit); } public int serverPort() { diff --git a/src/main/java/net/openhft/chronicle/hash/ThrottlingConfig.java b/src/main/java/net/openhft/chronicle/hash/ThrottlingConfig.java index 30b177a26..dbb466e2d 100644 --- a/src/main/java/net/openhft/chronicle/hash/ThrottlingConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/ThrottlingConfig.java @@ -103,7 +103,7 @@ public class ThrottlingConfig { * @return maximum bits per the given time unit, or {@code 0} if there is no throttling */ public final long throttling(TimeUnit perUnit) { - return throttlePerUnit().convert(throttle(), perUnit); + return throttlePerUnit.convert(throttle, perUnit); } @@ -115,7 +115,7 @@ public class ThrottlingConfig { * @return the bucketing interval for throttling in the given time units */ public long bucketInterval(TimeUnit unit) { - return unit.convert(bucketInterval(), bucketIntervalUnit()); + return unit.convert(bucketInterval, bucketIntervalUnit); } /** @@ -129,26 +129,7 @@ public class ThrottlingConfig { public ThrottlingConfig bucketInterval(long throttleBucketInterval, TimeUnit unit) { if (throttleBucketInterval <= 0L) throw new IllegalArgumentException(); - return create(throttle(), throttlePerUnit(), throttleBucketInterval, unit); - } - - - long throttle() { - return throttle; - } - - - TimeUnit throttlePerUnit() { - return throttlePerUnit; - } - - - public long bucketInterval() { - return bucketInterval; - } - - public TimeUnit bucketIntervalUnit() { - return bucketIntervalUnit; + return create(throttle, throttlePerUnit, throttleBucketInterval, unit); } @Override @@ -168,10 +149,10 @@ public class ThrottlingConfig { } if (o instanceof ThrottlingConfig) { ThrottlingConfig that = (ThrottlingConfig) o; - return (this.throttle == that.throttle()) - && (this.throttlePerUnit.equals(that.throttlePerUnit())) - && (this.bucketInterval == that.bucketInterval()) - && (this.bucketIntervalUnit.equals(that.bucketIntervalUnit())); + return (this.throttle == that.throttle) + && (this.throttlePerUnit.equals(that.throttlePerUnit)) + && (this.bucketInterval == that.bucketInterval) + && (this.bucketIntervalUnit.equals(that.bucketIntervalUnit)); } return false; } diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 350b80f79..e63616205 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -43,7 +43,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; /** * @author Rob Austin. */ -public abstract class AbstractChannelReplicator implements Closeable { +abstract class AbstractChannelReplicator implements Closeable { public static final int BITS_IN_A_BYTE = 8; diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java index 3e48fcf87..b464fb929 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -265,7 +265,7 @@ public abstract class AbstractChronicleMapBuilder 0) return entryAndValueAlignment().alignSize(entrySize); int size = metaDataBytes; @@ -881,6 +881,17 @@ public abstract class AbstractChronicleMapBuilder 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"); + } + } Closeable token = replicator.applyTo(this, result, result, map); if (replicators.size() == 1 && token.getClass() == UdpReplicator.class) { LOG.warn( diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 4c7daa078..97b73ad4e 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -41,6 +41,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import static java.nio.channels.SelectionKey.*; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT; /** @@ -93,11 +94,9 @@ public class TcpReplicator extends AbstractChannelReplicator implements Closeabl this.statelessServerConnector = statelessServerConnector; final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig(); - long throttleBucketInterval = TimeUnit.MILLISECONDS.convert(throttlingConfig - .bucketInterval(), throttlingConfig.bucketIntervalUnit()); + long throttleBucketInterval = throttlingConfig.bucketInterval(MILLISECONDS); - heartBeatIntervalMillis = TimeUnit.MILLISECONDS.convert(replicationConfig.heartBeatInterval(), - replicationConfig.heartBeatIntervalUnit()); + heartBeatIntervalMillis = replicationConfig.heartBeatInterval(MILLISECONDS); selectorTimeout = Math.min(heartBeatIntervalMillis, throttleBucketInterval);