From a83b57af9629ce65b51d6138a5c25fcac07817c0 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 12 Dec 2014 11:46:39 +0000 Subject: [PATCH] HCOLL-246 added test for - Tcp Replication fails with large values or entries sizes that don't " + "included internal replication bytes --- .../chronicle/map/CHMUseCasesTest.java | 2 + .../map/TcpReplicationLargeEntryTest.java | 156 ++++++++++++++++++ .../chronicle/map/TcpReplicationSoakTest.java | 52 +++--- 3 files changed, 178 insertions(+), 32 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/TcpReplicationLargeEntryTest.java diff --git a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java index 1ebd0e042..09053187e 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java +++ b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java @@ -687,6 +687,8 @@ public class CHMUseCasesTest { } } + + /** * For unsigned int -> unsigned int entries, the key can be on heap or off heap. */ diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationLargeEntryTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationLargeEntryTest.java new file mode 100644 index 000000000..c5e586958 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationLargeEntryTest.java @@ -0,0 +1,156 @@ +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.model.Byteable; +import net.openhft.lang.model.DataValueClasses; +import net.openhft.lang.values.IntValue; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.Closeable; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +/** + * @author Rob Austin. + */ +public class TcpReplicationLargeEntryTest { + + + private ChronicleMap map1; + private ChronicleMap map2; + private IntValue value; + static int s_port = 8010; + + @Before + public void setup() throws IOException { + value = DataValueClasses.newDirectReference(IntValue.class); + ((Byteable) value).bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0); + + final InetSocketAddress endpoint = new InetSocketAddress("localhost", s_port + 1); + + { + final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port, + endpoint).autoReconnectedUponDroppedConnection(true).name(" map1") + .heartBeatInterval(1, TimeUnit.SECONDS) + .packetSize(1024 * 64); + + + map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(Builder.SIZE + Builder.SIZE) + .actualSegments(1) + .replication((byte) 1, tcpConfig1) + .instance() + .name("map1") + .create(); + } + { + final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of + (s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2") + .heartBeatInterval(1, TimeUnit.SECONDS) + .packetSize(1024 * 64); + + map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(Builder.SIZE + Builder.SIZE) + .replication((byte) 2, tcpConfig2) + .instance() + .name("map2") + .create(); + + } + s_port += 2; + } + + + @After + public void tearDown() throws InterruptedException { + + for (final Closeable closeable : new Closeable[]{map1, map2}) { + try { + closeable.close(); + } catch (Exception e) { + e.printStackTrace(); + } + } + + System.gc(); + } + + Set threads; + + @Before + public void sampleThreads() { + threads = Thread.getAllStackTraces().keySet(); + } + + @After + public void checkThreadsShutdown() { + StatelessClientTest.checkThreadsShutdown(threads); + } + + + @Test + public void testSoakTestWithRandomData() throws IOException, InterruptedException { + + System.out.print("SoakTesting "); + for (int j = 1; j < 2 * Builder.SIZE; j++) { + if (j % 100 == 0) + System.out.print("."); + Random rnd = new Random(j); + for (int i = 1; i < 10; i++) { + final int select = rnd.nextInt(2); + final ChronicleMap map = select > 0 ? map1 : map2; + + if (rnd.nextBoolean()) { + map.put(rnd.nextInt(Builder.SIZE), "test" + j); + } else { + map.remove(rnd.nextInt(Builder.SIZE)); + } + } + } + Thread.sleep(1000); + System.out.println("\nwaiting till equal"); + + waitTillEqual(10000); + + Assert.assertEquals(map1, map2); + + } + + + private void waitTillEqual(final int timeOutMs) throws InterruptedException { + + Map map1UnChanged = new HashMap(); + Map map2UnChanged = new HashMap(); + + int numberOfTimesTheSame = 0; + for (int t = 0; t < timeOutMs + 100; t++) { + if (map1.equals(map2)) { + if (map1.equals(map1UnChanged) && map2.equals(map2UnChanged)) { + numberOfTimesTheSame++; + } else { + numberOfTimesTheSame = 0; + map1UnChanged = new HashMap(map1); + map2UnChanged = new HashMap(map2); + } + Thread.sleep(1); + if (numberOfTimesTheSame == 100) { + System.out.println("same"); + break; + } + + } + Thread.sleep(1); + } + } +} + diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index 0617e8298..2526f9126 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -1,20 +1,15 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; -import net.openhft.lang.io.ByteBufferBytes; -import net.openhft.lang.model.Byteable; -import net.openhft.lang.model.DataValueClasses; -import net.openhft.lang.values.IntValue; -import org.junit.After; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Test; +import org.junit.*; import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; -import java.nio.ByteBuffer; -import java.util.*; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; import java.util.concurrent.TimeUnit; /** @@ -25,13 +20,19 @@ public class TcpReplicationSoakTest { private ChronicleMap map1; private ChronicleMap map2; - private IntValue value; + static int s_port = 8010; + int valueSize = 1000000; + + char[] value = new char[valueSize]; + @Before public void setup() throws IOException { - value = DataValueClasses.newDirectReference(IntValue.class); - ((Byteable) value).bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0); + Arrays.fill(value, 'X'); + + int keySize = 4; + int entrySize = keySize + valueSize; final InetSocketAddress endpoint = new InetSocketAddress("localhost", s_port + 1); @@ -43,7 +44,7 @@ public class TcpReplicationSoakTest { map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(Builder.SIZE + Builder.SIZE) + .entries(entrySize) .actualSegments(1) .replication((byte) 1, tcpConfig1) .instance() @@ -57,7 +58,7 @@ public class TcpReplicationSoakTest { .packetSize(1024 * 64); map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(Builder.SIZE + Builder.SIZE) + .entries(entrySize) .replication((byte) 2, tcpConfig2) .instance() .name("map2") @@ -94,26 +95,13 @@ public class TcpReplicationSoakTest { StatelessClientTest.checkThreadsShutdown(threads); } - + @Ignore("HCOLL-246 Tcp Replication fails with large values or entries sizes that don't " + + "included internal replication bytes") @Test - public void testSoakTestWithRandomData() throws IOException, InterruptedException { + public void testLargeValues() throws IOException, InterruptedException { - System.out.print("SoakTesting "); - for (int j = 1; j < 2 * Builder.SIZE; j++) { - if (j % 100 == 0) - System.out.print("."); - Random rnd = new Random(j); - for (int i = 1; i < 10; i++) { - final int select = rnd.nextInt(2); - final ChronicleMap map = select > 0 ? map1 : map2; + map1.put(1, new String(value)); - if (rnd.nextBoolean()) { - map.put(rnd.nextInt(Builder.SIZE), "test" + j); - } else { - map.remove(rnd.nextInt(Builder.SIZE)); - } - } - } Thread.sleep(1000); System.out.println("\nwaiting till equal");