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()); }