From ae26c293c77847f180b23ac8bf0974ffc16e4de6 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 21 Nov 2014 11:04:26 +0000 Subject: [PATCH] HCOLL-134 changed the default so that is does not autoReconnectedUponDroppedConnection, however it better if most of the test cases have this can you can guarantee that the server is ready to accept TCP/IP before the client. --- .../TcpTransportAndNetworkConfig.java | 26 ++++++++++++++++--- .../openhft/chronicle/map/NodeDiscovery.java | 16 +++++++----- .../openhft/chronicle/map/TcpReplicator.java | 25 +++++++++++++----- .../net/openhft/chronicle/map/Builder.java | 2 +- ...tListenerTestWithTCPSocketReplication.java | 6 +++-- .../openhft/chronicle/map/MapByNameTest.java | 12 +++++---- ...cketReplication3wayPutReturnsNullTest.java | 3 ++- .../TCPSocketReplicationBootStrapTest.java | 25 +++++++++++++++--- .../map/TCPSocketReplicationNativeTest.java | 4 +-- .../map/TwoMapOnDifferentServersTest.java | 8 +++--- .../map/example/ChannelReplicationTest.java | 21 ++++++++++----- .../SetForTCPSocketReplication3wayTest.java | 2 +- 12 files changed, 107 insertions(+), 43 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java b/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java index 1fd6de86b..cfea4557b 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java @@ -20,7 +20,10 @@ package net.openhft.chronicle.hash.replication; import java.io.Serializable; import java.net.InetSocketAddress; -import java.util.*; +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.Collections.unmodifiableSet; @@ -40,6 +43,7 @@ public final class TcpTransportAndNetworkConfig implements Serializable { private final ThrottlingConfig throttlingConfig; private final long heartBeatInterval; private final TimeUnit heartBeatIntervalUnit; + private String name = "(unknown)"; private TcpTransportAndNetworkConfig(int serverPort, Set endpoints, int packetSize, @@ -70,7 +74,7 @@ public final class TcpTransportAndNetworkConfig implements Serializable { return new TcpTransportAndNetworkConfig( serverPort, unmodifiableSet(new HashSet(endpoints)), DEFAULT_PACKET_SIZE, - true, // autoReconnectedUponDroppedConnection + false, // autoReconnectedUponDroppedConnection ThrottlingConfig.noThrottling(), DEFAULT_HEART_BEAT_INTERVAL, DEFAULT_HEART_BEAT_INTERVAL_UNIT); @@ -131,6 +135,14 @@ public final class TcpTransportAndNetworkConfig implements Serializable { return packetSize; } + /** + * @param name the name used here is reflected in the Tcp Replicator and is typically used when debugging + */ + public TcpTransportAndNetworkConfig name(String name) { + this.name = name; + return this; + } + public TcpTransportAndNetworkConfig packetSize(int packetSize) { if (packetSize <= 0) throw new IllegalArgumentException(); @@ -181,7 +193,8 @@ public final class TcpTransportAndNetworkConfig implements Serializable { @Override public String toString() { return "TcpReplicationConfig{" + - "serverPort=" + serverPort + + "name=" + name() + + ", serverPort=" + serverPort + ", endpoints=" + endpoints + ", packetSize=" + packetSize + ", autoReconnectedUponDroppedConnection=" + autoReconnectedUponDroppedConnection + @@ -190,4 +203,11 @@ public final class TcpTransportAndNetworkConfig implements Serializable { ", heartBeatIntervalUnit=" + heartBeatIntervalUnit + '}'; } + + /** + * @return the name of this configuration + */ + public String name() { + return this.name; + } } diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index 44d547f64..f7b693816 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -172,7 +172,7 @@ public class NodeDiscovery { // the identifiers will come back to the callback on the nio thread, the update arrives at the // onRemoteNodeEvent - Thread.sleep(200); + Thread.sleep(1000); byte identifier; @@ -191,7 +191,7 @@ public class NodeDiscovery { final ProposedNodes proposedNodes = new ProposedNodes(ourAddressAndPort, identifier); - Thread.sleep(100); + Thread.sleep(1000); countDownLatch.set(new CountDownLatch(1)); @@ -245,16 +245,18 @@ public class NodeDiscovery { // add our identifier and host:port to the list of known identifiers knownNodes.add(ourAddressAndPort, identifier); - final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + /* final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(ourAddressAndPort.getPort(), - toInetSocketCollection(knownHostPorts0)) - - .heartBeatInterval(1, SECONDS); - + toInetSocketCollection(knownHostPorts0)); +*/ LOG.info("Using Remote identifier=" + identifier); nodeDiscoveryEventListenerAtomicReference.set(null); + + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8123, new + InetSocketAddress("192.168.1.253", 8123)); + final ReplicationHub replicationHub = ReplicationHub.builder() .maxEntrySize(10 * 1024) .tcpTransportAndNetwork(tcpConfig) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 4a262a53a..fb5bee879 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -79,6 +79,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private final @Nullable RemoteNodeValidator remoteNodeValidator; + private final String name; private long selectorTimeout; @@ -115,7 +116,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable this.replicationConfig = replicationConfig; this.remoteNodeValidator = remoteNodeValidator; - + this.name = replicationConfig.name(); start(); } @@ -159,11 +160,17 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable if (!key.isValid()) continue; - if (key.isAcceptable()) + if (key.isAcceptable()) { + if (LOG.isDebugEnabled()) + LOG.info("onAccept - " + name); onAccept(key); + } - if (key.isConnectable()) + if (key.isConnectable()) { + if (LOG.isDebugEnabled()) + LOG.debug("onConnect - " + name); onConnect(key); + } if (key.isReadable()) onRead(key, approxTime); @@ -206,6 +213,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } catch (Exception e) { LOG.error("", e); } finally { + if (LOG.isDebugEnabled()) + LOG.debug("closing name=" + name); if (!isClosed) { close(); } @@ -593,11 +602,13 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable int len = attached.entryReader.readSocketToBuffer(socketChannel); if (len == -1) { - socketChannel.shutdownInput(); - socketChannel.shutdownOutput(); - //todo we should call close, but if we do it breaks the bootstrap test case, further - // investigation required + if (replicationConfig.autoReconnectedUponDroppedConnection()) { + AbstractConnector connector = attached.connector; + if (connector != null) + connector.connectLater(); + } else + socketChannel.close(); return; } diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java index dd0773d38..921f7405b 100644 --- a/src/test/java/net/openhft/chronicle/map/Builder.java +++ b/src/test/java/net/openhft/chronicle/map/Builder.java @@ -73,7 +73,7 @@ public class Builder { final int serverPort, final InetSocketAddress... endpoints) throws IOException { TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(serverPort, Arrays.asList(endpoints)) - .heartBeatInterval(1L, TimeUnit.SECONDS); + .heartBeatInterval(1L, TimeUnit.SECONDS).autoReconnectedUponDroppedConnection(true); return ChronicleMapBuilder.of(kClass, vClass) .replication(identifier, tcpConfig); } diff --git a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java index 49beb7104..af9848ef4 100644 --- a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java +++ b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java @@ -238,7 +238,8 @@ public class EventListenerTestWithTCPSocketReplication { } private ChronicleMapBuilder clientBuilder() { - TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8077); + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8077) + .autoReconnectedUponDroppedConnection(true); return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) .replication((byte) 1, tcpConfig) @@ -257,7 +258,8 @@ public class EventListenerTestWithTCPSocketReplication { } private ChronicleMapBuilder serverBuilder() { - TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8076, new InetSocketAddress("localhost", 8077)); + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8076, new InetSocketAddress("localhost", 8077)) + .autoReconnectedUponDroppedConnection(true); return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) .replication((byte) 2, tcpConfig); diff --git a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java index 196d0b158..a4453f785 100644 --- a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java +++ b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java @@ -94,23 +94,25 @@ public class MapByNameTest { final ReplicationHubFindByName mapByName = nodeDiscovery.mapByName(); Thread.sleep(2000); - ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence + ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence .class, CharSequence.class) .minSegments(2) .entries(2) - .name("myMap5") + .name("myMap6") .removeReturnsNull(true); - ChronicleMap map = mapByName.create(builder); - map.put("hello", "world"); + ChronicleMap map = mapByName.create(builder); + map.put("hello", "world"); - // final ChronicleMap map = mapByName.from("myMap5"); + // final ChronicleMap map = mapByName.from("myMap5"); // allow time for replication Thread.sleep(2000); System.out.print(map); + map.close(); + } } diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication3wayPutReturnsNullTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication3wayPutReturnsNullTest.java index 3af0a9cc2..932f1fc55 100755 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication3wayPutReturnsNullTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication3wayPutReturnsNullTest.java @@ -48,7 +48,8 @@ public class TCPSocketReplication3wayPutReturnsNullTest { final byte identifier, final int serverPort, final InetSocketAddress... endpoints) throws IOException { - TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(serverPort, Arrays.asList(endpoints)); + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(serverPort, Arrays.asList(endpoints)) + .autoReconnectedUponDroppedConnection(true); return (T) ChronicleMapBuilder.of(Integer.class, CharSequence.class) .putReturnsNull(true) diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java index 041c066b4..69f4937fe 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.After; import org.junit.Test; @@ -25,6 +26,8 @@ import java.io.Closeable; import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; +import java.util.Arrays; +import java.util.concurrent.TimeUnit; import static net.openhft.chronicle.map.Builder.getPersistenceFile; import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder; @@ -87,12 +90,22 @@ public class TCPSocketReplicationBootStrapTest { @Test public void testBootstrapAndHeartbeat() throws IOException, InterruptedException { - map1 = newTcpSocketShmIntString((byte) 1, 8068, new InetSocketAddress("localhost", 8067)); + TcpTransportAndNetworkConfig map1Config = TcpTransportAndNetworkConfig.of(8068, Arrays.asList(new InetSocketAddress("localhost", 8067))) + .heartBeatInterval(1L, TimeUnit.SECONDS).name("map1").autoReconnectedUponDroppedConnection + (true); + + map1 = (ReplicatedChronicleMap) ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replication((byte) 1, map1Config).create(); File persistenceFile = getPersistenceFile(); + + TcpTransportAndNetworkConfig map2Config = TcpTransportAndNetworkConfig.of(8067) + .heartBeatInterval(1L, TimeUnit.SECONDS).name("map2"); + final ReplicatedChronicleMap map2a = (ReplicatedChronicleMap) - newTcpSocketShmBuilder(Integer.class, CharSequence.class, (byte) 2, 8067).createPersistedTo(persistenceFile); + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replication((byte) 2, map2Config).createPersistedTo(persistenceFile); map2a.put(10, "EXAMPLE-10"); // this will be the last time that map1 go an update from map2 @@ -106,6 +119,7 @@ public class TCPSocketReplicationBootStrapTest { map2a.close(); + { // restart map 2 but does not connect it to map1 final ChronicleMap map2b = ChronicleMapBuilder.of(Integer.class, @@ -116,10 +130,13 @@ public class TCPSocketReplicationBootStrapTest { } // now restart map2a and doConnect it to map1, map1 should bootstrap the missing entry - map2 = newTcpSocketShmBuilder(Integer.class, CharSequence.class, (byte) 2, 8067).createPersistedTo(persistenceFile); + TcpTransportAndNetworkConfig tcpConfigNewMap2 = TcpTransportAndNetworkConfig.of(8067) + .heartBeatInterval(1L, TimeUnit.SECONDS).name("newMap2"); + map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replication((byte) 2, tcpConfigNewMap2).createPersistedTo(persistenceFile); // add data into it - waitTillEqual(20000); + waitTillEqual(20000000); assertEquals("ADDED WHEN DISCONNECTED TO MAP1", map1.get(11)); diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationNativeTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationNativeTest.java index 5b6e479fd..a2679b4fe 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationNativeTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationNativeTest.java @@ -51,13 +51,13 @@ public class TCPSocketReplicationNativeTest { InetSocketAddress endpoint = new InetSocketAddress("localhost", 8077); TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(8076, endpoint) - .heartBeatInterval(1L, TimeUnit.SECONDS); + .heartBeatInterval(1L, TimeUnit.SECONDS).autoReconnectedUponDroppedConnection(true); map1 = ChronicleMapBuilder.of(Integer.class, LongValue.class) .replication((byte) 1, tcpConfig1).create(); TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of(8077) - .heartBeatInterval(1L, TimeUnit.SECONDS); + .heartBeatInterval(1L, TimeUnit.SECONDS).autoReconnectedUponDroppedConnection(true); map2 = ChronicleMapBuilder.of(Integer.class, LongValue.class) .replication((byte) 2, tcpConfig2).create(); diff --git a/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServersTest.java b/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServersTest.java index 484c8c769..b3059df03 100644 --- a/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServersTest.java +++ b/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServersTest.java @@ -45,17 +45,19 @@ public class TwoMapOnDifferentServersTest { final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8076, new InetSocketAddress("localhost", 8077)) - .heartBeatInterval(1, SECONDS); + .heartBeatInterval(1, SECONDS).autoReconnectedUponDroppedConnection(true); map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000) .replication((byte) 1, tcpConfig).create(); + TcpTransportAndNetworkConfig config2 = TcpTransportAndNetworkConfig.of(8077) + .heartBeatInterval(1, SECONDS).autoReconnectedUponDroppedConnection(true); + map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000) - .replication((byte) 2, TcpTransportAndNetworkConfig.of(8077) - .heartBeatInterval(1, SECONDS)).create(); + .replication((byte) 2, config2).create(); } @After diff --git a/src/test/java/net/openhft/chronicle/map/example/ChannelReplicationTest.java b/src/test/java/net/openhft/chronicle/map/example/ChannelReplicationTest.java index c185ee069..30b7bf2ac 100644 --- a/src/test/java/net/openhft/chronicle/map/example/ChannelReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/example/ChannelReplicationTest.java @@ -62,7 +62,7 @@ public class ChannelReplicationTest { TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(8086, new InetSocketAddress("localhost", 8087)) - .heartBeatInterval(1, SECONDS); + .heartBeatInterval(1, SECONDS).autoReconnectedUponDroppedConnection(true); hubOnServer1 = ReplicationHub.builder() @@ -96,7 +96,8 @@ public class ChannelReplicationTest { byte identifier = (byte) 2; TcpTransportAndNetworkConfig tcpConfig = - TcpTransportAndNetworkConfig.of(8087).heartBeatInterval(1, SECONDS); + TcpTransportAndNetworkConfig.of(8087).heartBeatInterval(1, SECONDS) + .autoReconnectedUponDroppedConnection(true); hubOnServer2 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) @@ -166,7 +167,7 @@ public class ChannelReplicationTest { TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(8086) - .heartBeatInterval(1, SECONDS); + .heartBeatInterval(1, SECONDS).autoReconnectedUponDroppedConnection(true); hubOnServer1 = ReplicationHub.builder() @@ -201,7 +202,9 @@ public class ChannelReplicationTest { byte identifier = (byte) 6; TcpTransportAndNetworkConfig tcpConfig = - TcpTransportAndNetworkConfig.of(8087, new InetSocketAddress("localhost", 8086)).heartBeatInterval(1, SECONDS); + TcpTransportAndNetworkConfig.of(8087, new InetSocketAddress("localhost", 8086)) + .heartBeatInterval(1, SECONDS) + .autoReconnectedUponDroppedConnection(true); ReplicationHub hubOnServer = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) @@ -224,7 +227,9 @@ public class ChannelReplicationTest { byte identifier = (byte) 3; TcpTransportAndNetworkConfig tcpConfig = - TcpTransportAndNetworkConfig.of(8088, new InetSocketAddress("localhost", 8086)).heartBeatInterval(1, SECONDS); + TcpTransportAndNetworkConfig.of(8088, new InetSocketAddress("localhost", 8086)) + .heartBeatInterval(1, SECONDS) + .autoReconnectedUponDroppedConnection(true); ReplicationHub hubOnServer3 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) @@ -290,7 +295,8 @@ public class ChannelReplicationTest { TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(8086, new InetSocketAddress("localhost", 8087)) - .heartBeatInterval(1, SECONDS); + .heartBeatInterval(1, SECONDS) + .autoReconnectedUponDroppedConnection(true); hubOnServer1 = ReplicationHub.builder() @@ -324,7 +330,8 @@ public class ChannelReplicationTest { byte identifier = (byte) 2; TcpTransportAndNetworkConfig tcpConfig = - TcpTransportAndNetworkConfig.of(8087).heartBeatInterval(1, SECONDS); + TcpTransportAndNetworkConfig.of(8087).heartBeatInterval(1, SECONDS) + .autoReconnectedUponDroppedConnection(true); hubOnServer2 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) diff --git a/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplication3wayTest.java b/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplication3wayTest.java index 0f685881a..5346a9731 100644 --- a/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplication3wayTest.java +++ b/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplication3wayTest.java @@ -54,7 +54,7 @@ public class SetForTCPSocketReplication3wayTest { final InetSocketAddress... endpoints) throws IOException { return (T) ChronicleSetBuilder.of(Integer.class) .replication(identifier, TcpTransportAndNetworkConfig.of(serverPort, asList(endpoints)) - .heartBeatInterval(1L, SECONDS)) + .heartBeatInterval(1L, SECONDS).autoReconnectedUponDroppedConnection(true)) .createPersistedTo(getPersistenceFile()); }