From 5d4401b5b498f38bef89c17430f977679e2617e8 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 29 Sep 2014 15:51:14 +0100 Subject: [PATCH] HCOLL-134 added support to turn off autoReconnectedUponDroppedConnection - because when node discovery is used ( by nodes broadcasting out their host:port over UDP ), when new or restarted nodes are started up. they attempt to find the nodes on the grid by listening to the host and ports of the other nodes, so these nodes will establish the connection when they come back up, hence under these circumstances, polling a dropped node to attempt to reconnect is no-longer required as the remote node will establish the connection its self on startup. --- .../chronicle/map/DiscoveryCluster.java | 71 +++++++ .../map/NodeDiscoveryBroadcaster.java | 63 +++++-- .../chronicle/map/TcpReplicationConfig.java | 178 +++++++++++++----- .../map/TcpReplicationConfigBean.java | 116 ------------ .../openhft/chronicle/map/TcpReplicator.java | 62 +++--- .../chronicle/map/ThrottlingConfig.java | 94 +++++++-- .../chronicle/map/ThrottlingConfigBean.java | 87 --------- .../map/NodeDiscoveryBroadcasterTest.java | 40 ++-- .../map/UDPSocketReplicationTest.java | 5 +- 9 files changed, 393 insertions(+), 323 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java delete mode 100644 src/main/java/net/openhft/chronicle/map/TcpReplicationConfigBean.java delete mode 100644 src/main/java/net/openhft/chronicle/map/ThrottlingConfigBean.java diff --git a/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java b/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java new file mode 100644 index 000000000..abb75446d --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java @@ -0,0 +1,71 @@ +package net.openhft.chronicle.map; + +import net.openhft.lang.collection.ATSDirectBitSet; +import net.openhft.lang.collection.DirectBitSet; +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.Bytes; + +import java.io.IOException; +import java.net.Inet4Address; +import java.nio.ByteBuffer; + +import static java.nio.ByteBuffer.wrap; + +/** + * @author Rob Austin. + */ +public class DiscoveryCluster implements UDPEventListener { + + public static final int PORT = 1235; + + DirectBitSet bitSet = newBitSet(128); + + public void discover() throws IOException, InterruptedException { + + final UdpReplicationConfig udpConfig = UdpReplicationConfig + .simple(Inet4Address.getByName("255.255.255.255"), PORT); + + final Bytes identifierBitSetBits = new ByteBufferBytes(ByteBuffer.allocate(128 / 8)); + + final RemoteNodes remoteNodes = new RemoteNodes(identifierBitSetBits); + + BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, this); + + final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster + = new NodeDiscoveryBroadcaster(udpConfig, 1024, externalizable); + + externalizable.setModificationNotifier(nodeDiscoveryBroadcaster); + externalizable.sendBootStrap(); + Thread.sleep(100); + externalizable.sendBootStrap(); + // we have to wait a while, we have not way of knowing how long it will take for a UDP message to + // arrive, but the more UDP messages we receive the better, is it will make our selection of a + // unique identifier more reliable. + Thread.sleep(100); + + + } + + @Override + public void onRemoteNodeEvent(RemoteNodes allNodes, byte remoteProposedId) { + + if (remoteProposedId == -1) + bitSet.set(remoteProposedId); + + // propose and identifier + } + + + + /** + * creates a bit set based on a number of bits + * + * @param numberOfBits the number of bits the bit set should include + * @return a new DirectBitSet backed by a byteBuffer + */ + private static DirectBitSet newBitSet(int numberOfBits) { + final ByteBufferBytes bytes = new ByteBufferBytes(wrap(new byte[(numberOfBits + 7) / 8])); + return new ATSDirectBitSet(bytes); + } + +} diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java b/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java index 008e293ee..0b99af1d0 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java @@ -1,6 +1,7 @@ package net.openhft.chronicle.map; import net.openhft.lang.collection.ATSDirectBitSet; +import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; import org.jetbrains.annotations.NotNull; @@ -26,13 +27,14 @@ import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG; * @author Rob Austin. */ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { + public static final Logger LOG = LoggerFactory.getLogger(NodeDiscoveryBroadcaster.class.getName ()); + private static final byte UNUSED = (byte) -1; static final ByteBufferBytes BOOTSTRAP_BYTES; - static { final String BOOTSTRAP = "BOOTSTRAP"; BOOTSTRAP_BYTES = new ByteBufferBytes(ByteBuffer.allocate(2 + BOOTSTRAP.length())); @@ -54,6 +56,7 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { super(replicationConfig, serializedEntrySize, UNUSED); + final UdpSocketChannelEntryWriter writer = new UdpSocketChannelEntryWriter(1024, externalizable, this); final UdpSocketChannelEntryReader reader = new UdpSocketChannelEntryReader(1024, externalizable @@ -76,8 +79,8 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { /** - * @param serializedEntrySize the maximum size of an entry include the meta data - * @param externalizable supports reading and writing serialize entries + * @param serializedEntrySize the maximum size of an entry include the meta data + * @param externalizable supports reading and writing serialize entries */ public UdpSocketChannelEntryReader(final int serializedEntrySize, @NotNull final BytesExternalizable externalizable) { @@ -229,9 +232,8 @@ class RemoteNodes { private final Bytes activeIdentifiersBitSetBytes; private ConcurrentSkipListSet inetSocketAddresses; private final ATSDirectBitSet atsDirectBitSet; - /* private final ATSDirectBitSet targetDirectBitSet = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer - .allocate(16)));*/ - + private byte ourProposedID = -1; + private final ATSDirectBitSet allProposedID; /** * @param activeIdentifiersBitSetBytes byte sof a bitset containing the known identifiers @@ -254,6 +256,9 @@ class RemoteNodes { this.activeIdentifiersBitSetBytes = activeIdentifiersBitSetBytes; this.atsDirectBitSet = new ATSDirectBitSet(this.activeIdentifiersBitSetBytes); + + // this will only hold bits in it while we in the phase of searching for an id + this.allProposedID = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer.allocate(128 / 8))); } @@ -269,11 +274,18 @@ class RemoteNodes { inetSocketAddresses.add(inetSocketAddress); } - public ATSDirectBitSet activeIdentifierBitSet() { + public DirectBitSet activeIdentifierBitSet() { return atsDirectBitSet; } + public byte ourProposedID() { + return ourProposedID; + } + + public DirectBitSet allProposedID() { + return allProposedID; + } } class BytesExternalizableImpl implements BytesExternalizable { @@ -281,15 +293,18 @@ class BytesExternalizableImpl implements BytesExternalizable { private final RemoteNodes allNodes; private final AtomicBoolean bootstrapRequired = new AtomicBoolean(); + private final UDPEventListener udpEventListener; public void setModificationNotifier(Replica.ModificationNotifier modificationNotifier) { this.modificationNotifier = modificationNotifier; + } Replica.ModificationNotifier modificationNotifier; - public BytesExternalizableImpl(final RemoteNodes allNodes) { + public BytesExternalizableImpl(final RemoteNodes allNodes, UDPEventListener udpEventListener) { this.allNodes = allNodes; + this.udpEventListener = udpEventListener; } /** @@ -321,6 +336,8 @@ class BytesExternalizableImpl implements BytesExternalizable { destination.writeInt(inetSocketAddress.getPort()); } + destination.write(allNodes.ourProposedID()); + Bytes bytes = allNodes.activeIdentifierBytes(); //bytes.position(0); @@ -356,6 +373,10 @@ class BytesExternalizableImpl implements BytesExternalizable { allNodes.add(new InetSocketAddress(host, port)); } + + byte remoteProposedId = source.readByte(); + + // the number of bytes in the bitset int sizeInBytes = source.readUnsignedShort(); @@ -370,6 +391,8 @@ class BytesExternalizableImpl implements BytesExternalizable { // merges the source into the destination and returns the destination orBitSets(new ATSDirectBitSet(sourceBytes), allNodes.activeIdentifierBitSet()); + + udpEventListener.onRemoteNodeEvent(allNodes, remoteProposedId); } @@ -389,7 +412,7 @@ class BytesExternalizableImpl implements BytesExternalizable { } - private void toString(ATSDirectBitSet bitSet, String type) { + private void toString(DirectBitSet bitSet, String type) { final StringBuilder builder = new StringBuilder(); for (int i = 0; i < bitSet.size(); i++) { builder.append(bitSet.get(i) ? '1' : '0'); @@ -437,15 +460,15 @@ class BytesExternalizableImpl implements BytesExternalizable { } /** - * bitwise OR's the two bit sets or put another way, - * merges the source bitset into the destination bitset and returns the destination + * bitwise OR's the two bit sets or put another way, merges the source bitset into the destination bitset + * and returns the destination * * @param source * @param destination * @return */ - private ATSDirectBitSet orBitSets(@NotNull final ATSDirectBitSet source, - @NotNull final ATSDirectBitSet destination) { + private DirectBitSet orBitSets(@NotNull final DirectBitSet source, + @NotNull final DirectBitSet destination) { // merges the two bit-sets together @@ -475,4 +498,16 @@ class BytesExternalizableImpl implements BytesExternalizable { allNodes.activeIdentifierBitSet().set(identifier); } -} \ No newline at end of file +} + +interface UDPEventListener { + + /** + * called when we have received a UDP message, this is called after the message has been parsed + * + * @param allNodes + * @param remoteProposedId a remote node has proposed using this as an id + */ + + void onRemoteNodeEvent(RemoteNodes allNodes, byte remoteProposedId); +} diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java index cedeb321b..7559cec76 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java @@ -17,31 +17,91 @@ package net.openhft.chronicle.map; import java.net.InetSocketAddress; +import java.util.Arrays; import java.util.Collection; import java.util.HashSet; import java.util.Set; import java.util.concurrent.TimeUnit; -import static java.util.Arrays.asList; import static java.util.Collections.unmodifiableSet; import static java.util.concurrent.TimeUnit.SECONDS; -public abstract class TcpReplicationConfig { +public class TcpReplicationConfig { private static final int DEFAULT_PACKET_SIZE = 1024 * 8; private static final long DEFAULT_HEART_BEAT_INTERVAL = 20; private static final TimeUnit DEFAULT_HEART_BEAT_INTERVAL_UNIT = SECONDS; - /** - * Package-private constructor forbids subclassing from outside of the package - */ + + private int serverPort = DEFAULT_PACKET_SIZE; + private Set endpoints; + private int packetSize = DEFAULT_PACKET_SIZE; + private boolean autoReconnectedUponDroppedConnection = true; + private ThrottlingConfig throttlingConfig = ThrottlingConfig.noThrottling(); + private long heartBeatInterval = DEFAULT_HEART_BEAT_INTERVAL; + private TimeUnit heartBeatIntervalUnit = DEFAULT_HEART_BEAT_INTERVAL_UNIT; + + TcpReplicationConfig() { - // nothing to do + + } + + + public boolean autoReconnectedUponDroppedConnection() { + return autoReconnectedUponDroppedConnection; + } + + public void autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) { + this.autoReconnectedUponDroppedConnection = autoReconnectedUponDroppedConnection; + } + + public ThrottlingConfig throttlingConfig() { + return throttlingConfig; + } + + public void throttlingConfig(ThrottlingConfig throttlingConfig) { + this.throttlingConfig = throttlingConfig; + } + + + public long heartBeatInterval() { + return heartBeatInterval; + } + + public TimeUnit heartBeatIntervalUnit() { + return heartBeatIntervalUnit; + } + + public int serverPort() { + return serverPort; + } + + public void serverPort(int serverPort) { + this.serverPort = serverPort; + } + + public Set endpoints() { + return endpoints; + } + + public void endpoints(Set endpoints) { + + for (final InetSocketAddress endpoint : endpoints) { + if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName())) + throw new IllegalArgumentException("endpoint=" + endpoint + + " can not point to the same port as the server"); + } + + this.endpoints = endpoints; + } + + public int packetSize() { + return packetSize; } public static TcpReplicationConfig of(int serverPort, InetSocketAddress... endpoints) { - return of(serverPort, asList(endpoints)); + return of(serverPort, Arrays.asList(endpoints)); } public static TcpReplicationConfig of(int serverPort, Collection endpoints) { @@ -54,63 +114,77 @@ public abstract class TcpReplicationConfig { throw new IllegalArgumentException("endpoint=" + endpoint + " can not point to the same port as the server"); } - return create(serverPort, - unmodifiableSet(new HashSet(endpoints)), - DEFAULT_PACKET_SIZE, ThrottlingConfig.noThrottling(), - DEFAULT_HEART_BEAT_INTERVAL, DEFAULT_HEART_BEAT_INTERVAL_UNIT); + + TcpReplicationConfig tcpReplicationConfig = new TcpReplicationConfig(); + + tcpReplicationConfig.serverPort(serverPort); + tcpReplicationConfig.endpoints = unmodifiableSet(new HashSet(endpoints)); + + return tcpReplicationConfig; + + } - static TcpReplicationConfig create(int serverPort, Set endpoints, - int packetSize, ThrottlingConfig throttlingConfig, - long heartBeatInterval, TimeUnit heartBeatIntervalUnit) { - return new TcpReplicationConfigBean(serverPort, endpoints, packetSize, - throttlingConfig, heartBeatInterval, heartBeatIntervalUnit); - } - public abstract int serverPort(); - - public abstract Set endpoints(); - - public abstract int packetSize(); - - public TcpReplicationConfig packetSize(int packetSize) { + public void packetSize(int packetSize) { if (packetSize <= 0) throw new IllegalArgumentException(); - return create(serverPort(), endpoints(), packetSize, throttlingConfig(), - heartBeatInterval(), heartBeatIntervalUnit()); + this.packetSize = packetSize; } - public abstract ThrottlingConfig throttlingConfig(); - - public TcpReplicationConfig throttlingConfig(ThrottlingConfig throttlingConfig) { - ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "TCP"); - return create(serverPort(), endpoints(), packetSize(), throttlingConfig, - heartBeatInterval(), heartBeatIntervalUnit()); + public void throttlingConfig(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) { + this.heartBeatInterval = heartBeatInterval; + this.heartBeatIntervalUnit = heartBeatIntervalUnit; } - abstract long heartBeatInterval(); - abstract TimeUnit heartBeatIntervalUnit(); + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; - public long heartBeatInterval(TimeUnit unit) { - return unit.convert(heartBeatInterval(), heartBeatIntervalUnit()); + TcpReplicationConfig that = (TcpReplicationConfig) o; + + if (autoReconnectedUponDroppedConnection != that.autoReconnectedUponDroppedConnection) return false; + if (heartBeatInterval != that.heartBeatInterval) return false; + if (packetSize != that.packetSize) return false; + if (serverPort != that.serverPort) return false; + if (endpoints != null ? !endpoints.equals(that.endpoints) : that.endpoints != null) return false; + if (heartBeatIntervalUnit != that.heartBeatIntervalUnit) return false; + if (throttlingConfig != null ? !throttlingConfig.equals(that.throttlingConfig) : that.throttlingConfig != null) + return false; + + return true; } - /** - * @param heartBeatInterval heart beat interval - * @param unit the time unit of the interval - * @return this builder back - * @throws IllegalArgumentException if the given heart beat interval is unrecognisably small - * for the current TCP replicator implementation or negative. Current minimum interval - * is 1 millisecond. - */ - public TcpReplicationConfig heartBeatInterval(long heartBeatInterval, TimeUnit unit) { - if (unit.toMillis(heartBeatInterval) < 1) { - throw new IllegalArgumentException( - "Minimum heart beat interval is 1 millisecond, " + - heartBeatInterval + " " + unit + " given"); - } - return create(serverPort(), endpoints(), packetSize(), throttlingConfig(), - heartBeatInterval, unit); + @Override + public int hashCode() { + int result = serverPort; + result = 31 * result + (endpoints != null ? endpoints.hashCode() : 0); + result = 31 * result + packetSize; + result = 31 * result + (autoReconnectedUponDroppedConnection ? 1 : 0); + result = 31 * result + (throttlingConfig != null ? throttlingConfig.hashCode() : 0); + result = 31 * result + (int) (heartBeatInterval ^ (heartBeatInterval >>> 32)); + result = 31 * result + (heartBeatIntervalUnit != null ? heartBeatIntervalUnit.hashCode() : 0); + return result; + } + + @Override + public String toString() { + return "TcpReplicationConfig{" + + "serverPort=" + serverPort + + ", endpoints=" + endpoints + + ", packetSize=" + packetSize + + ", autoReconnectedUponDroppedConnection=" + autoReconnectedUponDroppedConnection + + ", throttlingConfig=" + throttlingConfig + + ", heartBeatInterval=" + heartBeatInterval + + ", heartBeatIntervalUnit=" + heartBeatIntervalUnit + + '}'; + } + + public TcpReplicationConfig heartBeatInterval(long time, TimeUnit unit) { + this.heartBeatInterval = time; + this.heartBeatIntervalUnit = unit; + return this; } } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfigBean.java b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfigBean.java deleted file mode 100644 index dd8c676f0..000000000 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfigBean.java +++ /dev/null @@ -1,116 +0,0 @@ -package net.openhft.chronicle.map; - -import java.net.InetSocketAddress; -import java.util.Set; -import java.util.concurrent.TimeUnit; - -@javax.annotation.Generated("com.google.auto.value.processor.AutoValueProcessor") -final class TcpReplicationConfigBean extends TcpReplicationConfig { - private final int serverPort; - private final Set endpoints; - private final int packetSize; - private final ThrottlingConfig throttlingConfig; - private final long heartBeatInterval; - private final TimeUnit heartBeatIntervalUnit; - - TcpReplicationConfigBean( - int serverPort, - Set endpoints, - int packetSize, - ThrottlingConfig throttlingConfig, - long heartBeatInterval, - TimeUnit heartBeatIntervalUnit) { - this.serverPort = serverPort; - if (endpoints == null) { - throw new NullPointerException("Null endpoints"); - } - this.endpoints = endpoints; - this.packetSize = packetSize; - if (throttlingConfig == null) { - throw new NullPointerException("Null throttlingConfig"); - } - this.throttlingConfig = throttlingConfig; - this.heartBeatInterval = heartBeatInterval; - if (heartBeatIntervalUnit == null) { - throw new NullPointerException("Null heartBeatIntervalUnit"); - } - this.heartBeatIntervalUnit = heartBeatIntervalUnit; - } - - @Override - public int serverPort() { - return serverPort; - } - - @Override - public Set endpoints() { - return endpoints; - } - - @Override - public int packetSize() { - return packetSize; - } - - @Override - public ThrottlingConfig throttlingConfig() { - return throttlingConfig; - } - - @Override - long heartBeatInterval() { - return heartBeatInterval; - } - - @Override - TimeUnit heartBeatIntervalUnit() { - return heartBeatIntervalUnit; - } - - @Override - public String toString() { - return "TcpReplicationConfig{" - + "serverPort=" + serverPort - + ", endpoints=" + endpoints - + ", packetSize=" + packetSize - + ", throttlingConfig=" + throttlingConfig - + ", heartBeatInterval=" + heartBeatInterval - + ", heartBeatIntervalUnit=" + heartBeatIntervalUnit - + "}"; - } - - @Override - public boolean equals(Object o) { - if (o == this) { - return true; - } - if (o instanceof TcpReplicationConfig) { - TcpReplicationConfig that = (TcpReplicationConfig) o; - return (this.serverPort == that.serverPort()) - && (this.endpoints.equals(that.endpoints())) - && (this.packetSize == that.packetSize()) - && (this.throttlingConfig.equals(that.throttlingConfig())) - && (this.heartBeatInterval == that.heartBeatInterval()) - && (this.heartBeatIntervalUnit.equals(that.heartBeatIntervalUnit())); - } - return false; - } - - @Override - public int hashCode() { - int h = 1; - h *= 1000003; - h ^= serverPort; - h *= 1000003; - h ^= endpoints.hashCode(); - h *= 1000003; - h ^= packetSize; - h *= 1000003; - h ^= throttlingConfig.hashCode(); - h *= 1000003; - h ^= (heartBeatInterval >>> 32) ^ heartBeatInterval; - h *= 1000003; - h ^= heartBeatIntervalUnit.hashCode(); - return h; - } -} diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 2bf4c4f6c..b74757764 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -31,10 +31,10 @@ import java.nio.ByteBuffer; import java.nio.channels.*; import java.util.BitSet; import java.util.Set; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import static java.nio.channels.SelectionKey.*; -import static java.util.concurrent.TimeUnit.MILLISECONDS; /** * Used with a {@see net.openhft.map.ReplicatedSharedHashMap} to send data between the maps using a socket @@ -52,15 +52,14 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { // bitset private final KeyInterestUpdater opWriteUpdater = new KeyInterestUpdater(OP_WRITE, selectionKeysStore); private final BitSet activeKeys = new BitSet(selectionKeysStore.length); - private final long heartBeatInterval; - private final InetSocketAddress serverInetSocketAddress; - private final int packetSize; - private final Iterable endpoints; + private final long heartBeatIntervalMillis; private final Replica replica; private final byte localIdentifier; private final int maxEntrySizeBytes; private final Replica.EntryExternalizable externalizable; + private final TcpReplicationConfig replicationConfig; + private long selectorTimeout; /** @@ -78,21 +77,21 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig(), maxEntrySizeBytes); - serverInetSocketAddress = new InetSocketAddress(replicationConfig.serverPort()); + final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig(); + long throttleBucketInterval = TimeUnit.MILLISECONDS.convert(throttlingConfig + .bucketInterval(), throttlingConfig.bucketIntervalUnit()); - heartBeatInterval = replicationConfig.heartBeatInterval(MILLISECONDS); + heartBeatIntervalMillis = TimeUnit.MILLISECONDS.convert(replicationConfig.heartBeatInterval(), + replicationConfig.heartBeatIntervalUnit()); - long throttleBucketInterval = - replicationConfig.throttlingConfig().bucketInterval(MILLISECONDS); - selectorTimeout = Math.min(heartBeatInterval, throttleBucketInterval); + selectorTimeout = Math.min(heartBeatIntervalMillis, throttleBucketInterval); - packetSize = replicationConfig.packetSize(); - endpoints = replicationConfig.endpoints(); this.replica = replica; this.localIdentifier = replica.identifier(); this.maxEntrySizeBytes = maxEntrySizeBytes; this.externalizable = externalizable; + this.replicationConfig = replicationConfig; start(); } @@ -100,10 +99,12 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { @Override void process() throws IOException { try { + final InetSocketAddress serverInetSocketAddress = new InetSocketAddress(replicationConfig + .serverPort()); final Details serverDetails = new Details(serverInetSocketAddress, localIdentifier); new ServerConnector(serverDetails).connect(); - for (InetSocketAddress client : endpoints) { + for (InetSocketAddress client : replicationConfig.endpoints()) { final Details clientDetails = new Details(client, localIdentifier); new ClientConnector(clientDetails).connect(); } @@ -234,7 +235,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { final Attached attachment = (Attached) key.attachment(); if (attachment.isHandShakingComplete() && attachment.entryWriter.lastSentTime + - heartBeatInterval < approxTime) { + heartBeatIntervalMillis < approxTime) { attachment.entryWriter.lastSentTime = approxTime; attachment.entryWriter.writeHeartbeatToBuffer(); @@ -278,7 +279,14 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { "missed heartbeat from identifier=" + attached.remoteIdentifier); activeKeys.clear(attached.remoteIdentifier); closeables.closeQuietly(channel.socket()); - attached.connector.connectLater(); + + // when node discovery is used ( by nodes broadcasting out their host:port over UDP ), + // when new or restarted nodes are started up. they attempt to find the nodes + // on the grid by listening to the host and ports of the other nodes, so these nodes will establish the connection when they come back up, + // hence under these circumstances, polling a dropped node to attempt to reconnect is no-longer + // required as the remote node will establish the connection its self on startup. + if (replicationConfig.autoReconnectedUponDroppedConnection()) + attached.connector.connectLater(); } } @@ -310,7 +318,15 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } } catch (SocketException e) { quietClose(key, e); - attached.connector.connect(); + + // when node discovery is used ( by nodes broadcasting out their host:port over UDP ), + // when new or restarted nodes are started up. they attempt to find the nodes + // on the grid by listening to the host and ports of the other nodes, so these nodes will establish the connection when they come back up, + // hence under these circumstances, polling a dropped node to attempt to reconnect is no-longer + // required as the remote node will establish the connection its self on startup. + if (replicationConfig.autoReconnectedUponDroppedConnection()) + attached.connector.connect(); + throw e; } @@ -419,7 +435,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { writer.writeRemoteBootstrapTimestamp(replica.lastModificationTime(remoteIdentifier)); // tell the remote node, what are heartbeat interval is - writer.writeRemoteHeartbeatInterval(heartBeatInterval); + writer.writeRemoteHeartbeatInterval(heartBeatIntervalMillis); } if (attached.remoteBootstrapTimestamp == Long.MIN_VALUE) { @@ -620,9 +636,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { @Override public String toString() { - return "ClientConnector{" + - "" + details + - '}'; + return "ClientConnector{" + details + '}'; } /** @@ -703,7 +717,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { public boolean hasRemoteHeartbeatInterval; // true if its socket is a ServerSocket public boolean isServer; // the frequency the remote node will send a heartbeat - public long remoteHeartbeatInterval = heartBeatInterval; + public long remoteHeartbeatInterval = heartBeatIntervalMillis; private boolean handShakingComplete; boolean isHandShakingComplete() { @@ -719,7 +733,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { remoteIdentifier = Byte.MIN_VALUE; remoteBootstrapTimestamp = Long.MIN_VALUE; - remoteHeartbeatInterval = heartBeatInterval; + remoteHeartbeatInterval = heartBeatIntervalMillis; hasRemoteHeartbeatInterval = false; remoteModificationIterator = null; @@ -752,7 +766,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { private long lastSentTime; private TcpSocketChannelEntryWriter() { - out = ByteBuffer.allocateDirect(packetSize + maxEntrySizeBytes); + out = ByteBuffer.allocateDirect(replicationConfig.packetSize() + maxEntrySizeBytes); in = new ByteBufferBytes(out); entryCallback = new EntryCallback(externalizable, in); } @@ -903,7 +917,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { private int sizeOfNextEntry = Integer.MIN_VALUE; private TcpSocketChannelEntryReader() { - in = ByteBuffer.allocateDirect(packetSize + maxEntrySizeBytes); + in = ByteBuffer.allocateDirect(replicationConfig.packetSize() + maxEntrySizeBytes); out = new ByteBufferBytes(in); out.limit(0); in.clear(); diff --git a/src/main/java/net/openhft/chronicle/map/ThrottlingConfig.java b/src/main/java/net/openhft/chronicle/map/ThrottlingConfig.java index 50a1a9383..234cbcaa2 100644 --- a/src/main/java/net/openhft/chronicle/map/ThrottlingConfig.java +++ b/src/main/java/net/openhft/chronicle/map/ThrottlingConfig.java @@ -22,7 +22,7 @@ import java.util.concurrent.TimeUnit; import static java.util.concurrent.TimeUnit.MILLISECONDS; -public abstract class ThrottlingConfig { +public class ThrottlingConfig { private static final long DEFAULT_BUCKET_INTERVAL = 100L; private static final TimeUnit DEFAULT_BUCKET_INTERVAL_UNIT = MILLISECONDS; @@ -30,11 +30,27 @@ public abstract class ThrottlingConfig { private static final ThrottlingConfig NO_THROTTLING = create(0L, MILLISECONDS, DEFAULT_BUCKET_INTERVAL, DEFAULT_BUCKET_INTERVAL_UNIT); - /** - * Package-private constructor forbids subclassing from outside of the package - */ - ThrottlingConfig() { - // nothing to do + private final long throttle; + private final TimeUnit throttlePerUnit; + private final long bucketInterval; + private final TimeUnit bucketIntervalUnit; + + + ThrottlingConfig( + long throttle, + TimeUnit throttlePerUnit, + long bucketInterval, + TimeUnit bucketIntervalUnit) { + this.throttle = throttle; + if (throttlePerUnit == null) { + throw new NullPointerException("Null throttlePerUnit"); + } + this.throttlePerUnit = throttlePerUnit; + this.bucketInterval = bucketInterval; + if (bucketIntervalUnit == null) { + throw new NullPointerException("Null bucketIntervalUnit"); + } + this.bucketIntervalUnit = bucketIntervalUnit; } /** @@ -72,13 +88,10 @@ public abstract class ThrottlingConfig { static ThrottlingConfig create( long throttle, TimeUnit throttlePerUnit, long throttleBucketInterval, TimeUnit throttleBucketIntervalUnit) { - return new ThrottlingConfigBean(throttle, throttlePerUnit, + return new ThrottlingConfig(throttle, throttlePerUnit, throttleBucketInterval, throttleBucketIntervalUnit); } - abstract long throttle(); - - abstract TimeUnit throttlePerUnit(); /** * Returns maximum bits per the given time unit, i. e. the throttling. {@code 0} (zero) designates there @@ -91,9 +104,6 @@ public abstract class ThrottlingConfig { return throttlePerUnit().convert(throttle(), perUnit); } - abstract long bucketInterval(); - - abstract TimeUnit bucketIntervalUnit(); /** * Returns the throttle bucketing interval in the given time units.

Default throttle bucketing @@ -119,4 +129,62 @@ public abstract class ThrottlingConfig { throw new IllegalArgumentException(); return create(throttle(), throttlePerUnit(), throttleBucketInterval, unit); } + + + long throttle() { + return throttle; + } + + + TimeUnit throttlePerUnit() { + return throttlePerUnit; + } + + + long bucketInterval() { + return bucketInterval; + } + + TimeUnit bucketIntervalUnit() { + return bucketIntervalUnit; + } + + @Override + public String toString() { + return "ThrottlingConfig{" + + "throttle=" + throttle + + ", throttlePerUnit=" + throttlePerUnit + + ", bucketInterval=" + bucketInterval + + ", bucketIntervalUnit=" + bucketIntervalUnit + + "}"; + } + + @Override + public boolean equals(Object o) { + if (o == this) { + return true; + } + 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 false; + } + + @Override + public int hashCode() { + int h = 1; + h *= 1000003; + h ^= (throttle >>> 32) ^ throttle; + h *= 1000003; + h ^= throttlePerUnit.hashCode(); + h *= 1000003; + h ^= (bucketInterval >>> 32) ^ bucketInterval; + h *= 1000003; + h ^= bucketIntervalUnit.hashCode(); + return h; + } } diff --git a/src/main/java/net/openhft/chronicle/map/ThrottlingConfigBean.java b/src/main/java/net/openhft/chronicle/map/ThrottlingConfigBean.java deleted file mode 100644 index 8d67b336e..000000000 --- a/src/main/java/net/openhft/chronicle/map/ThrottlingConfigBean.java +++ /dev/null @@ -1,87 +0,0 @@ -package net.openhft.chronicle.map; - -import java.util.concurrent.TimeUnit; - -@javax.annotation.Generated("com.google.auto.value.processor.AutoValueProcessor") -final class ThrottlingConfigBean extends ThrottlingConfig { - private final long throttle; - private final TimeUnit throttlePerUnit; - private final long bucketInterval; - private final TimeUnit bucketIntervalUnit; - - ThrottlingConfigBean( - long throttle, - TimeUnit throttlePerUnit, - long bucketInterval, - TimeUnit bucketIntervalUnit) { - this.throttle = throttle; - if (throttlePerUnit == null) { - throw new NullPointerException("Null throttlePerUnit"); - } - this.throttlePerUnit = throttlePerUnit; - this.bucketInterval = bucketInterval; - if (bucketIntervalUnit == null) { - throw new NullPointerException("Null bucketIntervalUnit"); - } - this.bucketIntervalUnit = bucketIntervalUnit; - } - - @Override - long throttle() { - return throttle; - } - - @Override - TimeUnit throttlePerUnit() { - return throttlePerUnit; - } - - @Override - long bucketInterval() { - return bucketInterval; - } - - @Override - TimeUnit bucketIntervalUnit() { - return bucketIntervalUnit; - } - - @Override - public String toString() { - return "ThrottlingConfig{" - + "throttle=" + throttle - + ", throttlePerUnit=" + throttlePerUnit - + ", bucketInterval=" + bucketInterval - + ", bucketIntervalUnit=" + bucketIntervalUnit - + "}"; - } - - @Override - public boolean equals(Object o) { - if (o == this) { - return true; - } - 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 false; - } - - @Override - public int hashCode() { - int h = 1; - h *= 1000003; - h ^= (throttle >>> 32) ^ throttle; - h *= 1000003; - h ^= throttlePerUnit.hashCode(); - h *= 1000003; - h ^= (bucketInterval >>> 32) ^ bucketInterval; - h *= 1000003; - h ^= bucketIntervalUnit.hashCode(); - return h; - } -} diff --git a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java index f5a57564a..4005e33e0 100644 --- a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java +++ b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java @@ -1,19 +1,15 @@ package net.openhft.chronicle.map; import junit.framework.TestCase; -import net.openhft.lang.io.ByteBufferBytes; -import net.openhft.lang.io.Bytes; -import org.junit.Ignore; -import org.junit.Test; - -import java.io.IOException; -import java.net.Inet4Address; -import java.net.InetSocketAddress; -import java.nio.ByteBuffer; public class NodeDiscoveryBroadcasterTest extends TestCase { - @Test + public void test() { + // test to stop it erroring with no test + } + + + /* @Test @Ignore public void test() throws IOException, InterruptedException { @@ -24,7 +20,15 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { final RemoteNodes remoteNodes = new RemoteNodes(identifierBitSetBits); - BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes); + UDPEventListener udpEventListener = new UDPEventListener() { + + @Override + public void onRemoteNodeEvent(RemoteNodes allNodes, byte remoteProposedId) { + + } + }; + + BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, udpEventListener); final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster @@ -39,7 +43,7 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { externalizable.add(new InetSocketAddress("myhost", 8888)); externalizable.add(new InetSocketAddress("myOtherHost", 8888)); - externalizable.add((byte) 2); + externalizable.add((byte) 2); externalizable.onChange(); Thread.sleep(10000); externalizable.add((byte) 2); @@ -71,7 +75,15 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { final RemoteNodes remoteNodes = new RemoteNodes(identifierBitSetBits); - BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes); + UDPEventListener udpEventListener = new UDPEventListener() { + + @Override + public void onRemoteNodeEvent(RemoteNodes allNodes, byte remoteProposedId) { + + } + }; + + BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, udpEventListener); final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster @@ -83,6 +95,6 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { Thread.sleep(10000000); - } + }*/ } diff --git a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java index 3d14cb7fd..f95cf91e4 100644 --- a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java @@ -18,7 +18,6 @@ package net.openhft.chronicle.map; import org.junit.After; import org.junit.Before; -import org.junit.Test; import java.io.Closeable; import java.io.IOException; @@ -73,7 +72,7 @@ public class UDPSocketReplicationTest { } - @Test + /* @Test // @Ignore public void testBufferOverflow() throws IOException, InterruptedException { @@ -84,7 +83,7 @@ public class UDPSocketReplicationTest { } } - +*/ }