From a21265ef4a29df70e1651e5faf642d2a74e21ffc Mon Sep 17 00:00:00 2001 From: tolgakavukcuevam Date: Wed, 20 Jan 2016 16:15:43 +0200 Subject: [PATCH 1/5] ReplicationFixes --- .../openhft/chronicle/map/TcpReplicator.java | 3 +- .../TCPSocketReplicationBootStrapTest.java | 269 +++++++++++++++++- 2 files changed, 262 insertions(+), 10 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index ce442529a..5a55c4265 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -650,11 +650,12 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo LOG.debug("handshaking for localIdentifier=" + localIdentifier + "," + "remoteIdentifier=" + remoteIdentifier); - long timeStampOfLastMessage = replica.lastModificationTime(remoteIdentifier); + attached.remoteModificationIterator = replica.acquireModificationIterator(remoteIdentifier); attached.remoteModificationIterator.setModificationNotifier(attached); + long timeStampOfLastMessage = replica.lastModificationTime(remoteIdentifier); writer.writeRemoteBootstrapTimestamp(timeStampOfLastMessage); writer.writeServerVersion(); diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java index b61139887..921c92ef5 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java @@ -16,7 +16,9 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import org.jetbrains.annotations.Nullable; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -33,6 +35,7 @@ import static net.openhft.chronicle.map.Builder.getPersistenceFile; import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder; import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmIntString; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; /** * Test ReplicatedChronicleMap where the Replicated is over a TCP Socket @@ -60,7 +63,7 @@ public class TCPSocketReplicationBootStrapTest { long lastModificationTime; - // lets make sure that the message has got to map 1 + // lets make sure that the message has got to chronicleMap 1 do { lastModificationTime = map1.lastModificationTime((byte) 2); Thread.yield(); @@ -70,7 +73,7 @@ public class TCPSocketReplicationBootStrapTest { map2a.close(); { - // restart map 2 but don't doConnect it to map one + // restart chronicleMap 2 but don't doConnect it to chronicleMap one final ChronicleMap map2b = map2aBuilder.createPersistedTo(map2File); // add data into it @@ -89,6 +92,117 @@ public class TCPSocketReplicationBootStrapTest { persistenceFile.delete(); } + @Test + public void testReplication() throws IOException, InterruptedException { + + SingleChronicleHashReplication replicationHubForChannelIdMap = + SingleChronicleHashReplication.builder().bootstrapOnlyLocalEntries(true) + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5085,new InetSocketAddress[0])) + .createWithId((byte)1); + replicationHubForChannelIdMap.bootstrapOnlyLocalEntries(); + + ChronicleMap channelIdMap = ChronicleMapBuilder.of(java.lang.String.class, Integer.class) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap) + .create(); + + + channelIdMap.put("1",1); + SingleChronicleHashReplication replicationHubForChannelIdMap2 = + SingleChronicleHashReplication.builder().bootstrapOnlyLocalEntries(true) + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5086,new InetSocketAddress("localhost",5085))) + .createWithId((byte)2); + replicationHubForChannelIdMap.bootstrapOnlyLocalEntries(); + + ChronicleMap channelIdMap2 = ChronicleMapBuilder.of(java.lang.String.class, Integer.class) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap2) + .create(); + + + Thread.sleep(300); + + int map2Size = channelIdMap2.size(); + channelIdMap.close(); + channelIdMap2.close(); + assertEquals(1,map2Size); + } + + @Test + public void testReplicationWhileModifying() throws IOException, InterruptedException { + + SingleChronicleHashReplication replicationHubForChannelIdMap = + SingleChronicleHashReplication.builder().bootstrapOnlyLocalEntries(true) + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5085,new InetSocketAddress[0])) + .createWithId((byte)1); + + ChronicleMap channelIdMap = ChronicleMapBuilder.of(java.lang.String.class, Integer.class) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap) + .create(); + + int i; + for(i = 0;i < 500;i++){ + channelIdMap.put(Integer.toString(i),i); + } + MapReader reader = new MapReader(); + Thread t = new Thread(reader); + t.start(); + int j = 0; + while (j < 100){ + i = 400; + while (i < 500){ + channelIdMap.update(Integer.toString(i),i+1); + i++; + } + j++; + } + channelIdMap.close(); + reader.stop(); + int firstGatheredSize = reader.getFirstGatheredSize(); + assertEquals(0,reader.getFirstGatheredElement()); + assertEquals(500,firstGatheredSize); + } + + @Test + public void testReplicationWhileModifyingWithMapListener() throws IOException, InterruptedException { + + SingleChronicleHashReplication replicationHubForChannelIdMap = + SingleChronicleHashReplication.builder() + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5085,new InetSocketAddress[0])) + .createWithId((byte)1); + + ChronicleMap channelIdMap = ChronicleMapBuilder.of(java.lang.String.class, Integer.class) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap) + .create(); + + int i; + for(i = 0;i < 500;i++){ + channelIdMap.put(Integer.toString(i),i); + } + MapReaderWithListener reader = new MapReaderWithListener(); + Thread t = new Thread(reader); + t.start(); + int j = 0; + while (j < 100){ + i = 400; + while (i < 500){ + channelIdMap.update(Integer.toString(i),i+1); + i++; + } + j++; + } + Thread.sleep(10000); + channelIdMap.close(); + reader.stop(); + assertEquals(0,reader.getFirstGatheredElement()); + } + @Test public void testBootstrapAndHeartbeat() throws IOException, InterruptedException { @@ -113,7 +227,7 @@ public class TCPSocketReplicationBootStrapTest { long lastModificationTime; - // lets make sure that the message has got to map 1 + // lets make sure that the message has got to chronicleMap 1 do { lastModificationTime = map1.lastModificationTime((byte) 2); Thread.yield(); @@ -122,7 +236,7 @@ public class TCPSocketReplicationBootStrapTest { map2a.close(); { - // restart map 2 but does not connect it to map1 + // restart chronicleMap 2 but does not connect it to map1 final ChronicleMap map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class).replication((byte) 2).createPersistedTo(persistenceFile); // add data into it @@ -143,11 +257,13 @@ public class TCPSocketReplicationBootStrapTest { @After public void tearDown() { - for (final Closeable closeable : new Closeable[]{map1, map2}) { - try { - closeable.close(); - } catch (Exception e) { - e.printStackTrace(); + if(map1 != null && map2 != null) { + for (final Closeable closeable : new Closeable[]{map1, map2}) { + try { + closeable.close(); + } catch (Exception e) { + e.printStackTrace(); + } } } System.gc(); @@ -180,5 +296,140 @@ public class TCPSocketReplicationBootStrapTest { } } + private class MapReader implements Runnable { + + private volatile ChronicleMap chronicleMap; + private boolean stop = false; + private int firstGatheredSize = 0; + private Object firstGatheredElement; + + public MapReader() { + } + + public void stop() { + this.stop = true; + } + + public void run() { + + try { + SingleChronicleHashReplication replicationHubForChannelIdMap2 = + SingleChronicleHashReplication.builder() + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5086, new InetSocketAddress("localhost", 5085))) + .createWithId((byte) 2); + ChronicleMap channelIdMap2 = ChronicleMapBuilder.of(String.class, Integer.class) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap2) + .create(); + this.chronicleMap = channelIdMap2; + Object firstEl = 0; + while (!stop) { + int size = channelIdMap2.size(); + //System.out.println("Size=[" + size + "] on timestamp [" + Long.toString(System.currentTimeMillis()) + "]"); + if (size != 0 && this.firstGatheredSize == 0) { + this.firstGatheredSize = size; + firstEl = channelIdMap2.get("0"); + } + } + this.chronicleMap.close(); + firstGatheredElement = firstEl; + } catch (Exception e) { + e.printStackTrace(); + } + + } + + public int getFirstGatheredSize() { + return this.firstGatheredSize; + } + + public Object getFirstGatheredElement() { + return this.firstGatheredElement; + } + + } + + private class MapReaderWithListener implements Runnable { + + private volatile ChronicleMap chronicleMap; + private boolean stop = false; + private int firstGatheredSize = 0; + private Object firstGatheredElement; + private CacheMapListener mapListener; + + public MapReaderWithListener() { + } + + public void stop() { + while (mapListener.firstPuttedValue == null){ + this.stop = false; + } + this.stop = true; + } + + public void run() { + + try { + this.mapListener = new CacheMapListener<>(); + + SingleChronicleHashReplication replicationHubForChannelIdMap2 = + SingleChronicleHashReplication.builder() + .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(5086, new InetSocketAddress("localhost", 5085))) + .createWithId((byte) 2); + ChronicleMap channelIdMap2 = ChronicleMapBuilder.of(String.class, Integer.class).eventListener(this.mapListener) + .entries(Short.MAX_VALUE) + .instance() + .replicated(replicationHubForChannelIdMap2) + .create(); + this.chronicleMap = channelIdMap2; + while (!stop) { + channelIdMap2.size(); + } + this.firstGatheredElement = mapListener.firstPuttedValue; + this.chronicleMap.close(); + } catch (Exception e) { + e.printStackTrace(); + } + + } + + public int getFirstGatheredSize() { + return this.firstGatheredSize; + } + + public Object getFirstGatheredElement() { + return this.firstGatheredElement; + } + + } + + public class CacheMapListener extends MapEventListener { + + + private Object firstPuttedValue = null; + + public CacheMapListener() { + + } + + @Override + public void onPut(K key, V newValue, @Nullable V replacedValue, boolean replicationEvent, + boolean added, boolean hasValueChanged, byte identifier, + byte replacedIdentifier, long timeStamp, long replacedTimeStamp) { + if(this.firstPuttedValue == null){ + this.firstPuttedValue = newValue; + } + } + + @Override + public void onRemove(K key, V value, boolean replicationEvent, byte identifier, + byte replacedIdentifier, long timestamp, long replacedTimeStamp) { + + } + } + + + } From 917e0c4a31cb4f202b73f74e7cffc2cf702f0e51 Mon Sep 17 00:00:00 2001 From: tolgakavukcuevam Date: Wed, 20 Jan 2016 16:42:18 +0200 Subject: [PATCH 2/5] Change test assertions and close order of maps --- .../TCPSocketReplicationBootStrapTest.java | 21 ++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java index 921c92ef5..b7729af28 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java @@ -160,8 +160,8 @@ public class TCPSocketReplicationBootStrapTest { } j++; } - channelIdMap.close(); reader.stop(); + channelIdMap.close(); int firstGatheredSize = reader.getFirstGatheredSize(); assertEquals(0,reader.getFirstGatheredElement()); assertEquals(500,firstGatheredSize); @@ -198,8 +198,8 @@ public class TCPSocketReplicationBootStrapTest { j++; } Thread.sleep(10000); - channelIdMap.close(); reader.stop(); + channelIdMap.close(); assertEquals(0,reader.getFirstGatheredElement()); } @@ -307,6 +307,10 @@ public class TCPSocketReplicationBootStrapTest { } public void stop() { + this.stop = false; + while (firstGatheredSize == 0){ + this.stop = false; + } this.stop = true; } @@ -362,7 +366,8 @@ public class TCPSocketReplicationBootStrapTest { } public void stop() { - while (mapListener.firstPuttedValue == null){ + this.stop = false; + while (this.mapListener.firstPuttedValue == null){ this.stop = false; } this.stop = true; @@ -386,7 +391,7 @@ public class TCPSocketReplicationBootStrapTest { while (!stop) { channelIdMap2.size(); } - this.firstGatheredElement = mapListener.firstPuttedValue; + this.firstGatheredElement = this.mapListener.firstPuttedValue; this.chronicleMap.close(); } catch (Exception e) { e.printStackTrace(); @@ -417,9 +422,11 @@ public class TCPSocketReplicationBootStrapTest { public void onPut(K key, V newValue, @Nullable V replacedValue, boolean replicationEvent, boolean added, boolean hasValueChanged, byte identifier, byte replacedIdentifier, long timeStamp, long replacedTimeStamp) { - if(this.firstPuttedValue == null){ - this.firstPuttedValue = newValue; - } + + if(this.firstPuttedValue == null){ + this.firstPuttedValue = newValue; + } + } @Override From 2a9b27123189c44331d4bbebec119c94bf203b55 Mon Sep 17 00:00:00 2001 From: tolgakavukcuevam Date: Wed, 20 Jan 2016 17:12:30 +0200 Subject: [PATCH 3/5] Add more update to test to make sure case matches --- .../map/TCPSocketReplicationBootStrapTest.java | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java index b7729af28..72221ee09 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java @@ -189,12 +189,13 @@ public class TCPSocketReplicationBootStrapTest { Thread t = new Thread(reader); t.start(); int j = 0; - while (j < 100){ + while (j < 1000){ i = 400; while (i < 500){ channelIdMap.update(Integer.toString(i),i+1); i++; } + Thread.sleep(1); j++; } Thread.sleep(10000); @@ -359,17 +360,12 @@ public class TCPSocketReplicationBootStrapTest { private volatile ChronicleMap chronicleMap; private boolean stop = false; private int firstGatheredSize = 0; - private Object firstGatheredElement; private CacheMapListener mapListener; public MapReaderWithListener() { } public void stop() { - this.stop = false; - while (this.mapListener.firstPuttedValue == null){ - this.stop = false; - } this.stop = true; } @@ -391,7 +387,7 @@ public class TCPSocketReplicationBootStrapTest { while (!stop) { channelIdMap2.size(); } - this.firstGatheredElement = this.mapListener.firstPuttedValue; + ; this.chronicleMap.close(); } catch (Exception e) { e.printStackTrace(); @@ -404,7 +400,7 @@ public class TCPSocketReplicationBootStrapTest { } public Object getFirstGatheredElement() { - return this.firstGatheredElement; + return this.mapListener.firstPuttedValue; } } From 3148bb7209162fb5cd4f4bde4dec1d53c3845aa2 Mon Sep 17 00:00:00 2001 From: tolgakavukcuevam Date: Wed, 20 Jan 2016 17:15:29 +0200 Subject: [PATCH 4/5] Use longer update time and remove unecessary sleep --- .../chronicle/map/TCPSocketReplicationBootStrapTest.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java index 72221ee09..2c0458c3e 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java @@ -189,7 +189,7 @@ public class TCPSocketReplicationBootStrapTest { Thread t = new Thread(reader); t.start(); int j = 0; - while (j < 1000){ + while (j < 15000){ i = 400; while (i < 500){ channelIdMap.update(Integer.toString(i),i+1); @@ -198,7 +198,6 @@ public class TCPSocketReplicationBootStrapTest { Thread.sleep(1); j++; } - Thread.sleep(10000); reader.stop(); channelIdMap.close(); assertEquals(0,reader.getFirstGatheredElement()); From 30237f08a16ceaadbdcd088906693535672a3b02 Mon Sep 17 00:00:00 2001 From: tolgakavukcuevam Date: Thu, 21 Jan 2016 11:03:14 +0200 Subject: [PATCH 5/5] TCP Replicator timestamp of last Message --- src/main/java/net/openhft/chronicle/map/TcpReplicator.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 5a55c4265..ae5484edf 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -650,12 +650,12 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo LOG.debug("handshaking for localIdentifier=" + localIdentifier + "," + "remoteIdentifier=" + remoteIdentifier); - + long timeStampOfLastMessage = replica.lastModificationTime(remoteIdentifier); attached.remoteModificationIterator = replica.acquireModificationIterator(remoteIdentifier); attached.remoteModificationIterator.setModificationNotifier(attached); - long timeStampOfLastMessage = replica.lastModificationTime(remoteIdentifier); + writer.writeRemoteBootstrapTimestamp(timeStampOfLastMessage); writer.writeServerVersion();