From 7460074ac5ae45d3cb2c41ae76e65ccc5a8d1cbc Mon Sep 17 00:00:00 2001 From: leventov Date: Sat, 3 Oct 2015 05:06:39 +0300 Subject: [PATCH] Moved testFloodReplicatedMapWithDeletedEntries() test to a separate class from TcpReplicationSoakTest because they require different old deleted entries cleanup timeout configurations --- .../map/OldDeletedEntriesCleanupTest.java | 178 ++++++++++++++++++ .../chronicle/map/TcpReplicationSoakTest.java | 80 ++++---- 2 files changed, 217 insertions(+), 41 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupTest.java diff --git a/src/test/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupTest.java b/src/test/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupTest.java new file mode 100644 index 000000000..c58d757c5 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupTest.java @@ -0,0 +1,178 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.map; + +import com.google.common.collect.Maps; +import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; +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.*; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; + +public class OldDeletedEntriesCleanupTest { + + private ChronicleMap map1; + private ChronicleMap map2; + static int s_port = 8093; + + @Before + public void setup() throws IOException { + IntValue 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) + .heartBeatInterval(1, TimeUnit.SECONDS) + .tcpBufferSize(1024 * 64); + + + map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(Builder.SIZE + Builder.SIZE) + .removedEntryCleanupTimeout(1, TimeUnit.MILLISECONDS) + .actualSegments(1) + .averageValue("test" + 1000) + .replication(SingleChronicleHashReplication.builder() + .tcpTransportAndNetwork(tcpConfig1) + .name("map1") + .createWithId((byte) 1)) + .instance() + .name("map1") + .create(); + } + { + final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of + (s_port + 1).autoReconnectedUponDroppedConnection(true) + .heartBeatInterval(1, TimeUnit.SECONDS) + .tcpBufferSize(1024 * 64); + + map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(Builder.SIZE + Builder.SIZE) + .removedEntryCleanupTimeout(1, TimeUnit.MILLISECONDS) + .averageValue("test" + 1000) + .replication(SingleChronicleHashReplication.builder() + .tcpTransportAndNetwork(tcpConfig2) + .name("map2") + .createWithId((byte) 2)) + .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() { + ChannelReplicationTest.checkThreadsShutdown(threads); + } + + + @Test + public void testFloodReplicatedMapWithDeletedEntries() throws InterruptedException { + try { + Random r = ThreadLocalRandom.current(); + NavigableSet put = new TreeSet<>(); + for (int i = 0; i < Builder.SIZE * 10; i++) { + if (r.nextBoolean() || put.isEmpty()) { + int key = r.nextInt(); + map1.put(key, "test"); + put.add(key); + } else { + Integer key = put.pollFirst(); + map1.remove(key); + } +// Thread.sleep(0, 1000); + } + System.out.println("\nwaiting till equal"); + + waitTillEqual(15000); + + if (!map1.equals(map2)) + Assert.assertEquals(Maps.difference(map1, map2).toString(), map1, map2); + } finally { + map1.close(); + map2.close(); + } + } + + private void waitTillEqual(final int timeOutMs) throws InterruptedException { + + Map map1UnChanged = new HashMap(); + Map map2UnChanged = new HashMap(); + + int numberOfTimesTheSame = 0; + long startTime = System.currentTimeMillis(); + 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 == 10) { + System.out.println("same"); + break; + } + + } + Thread.sleep(1); + if (System.currentTimeMillis() - startTime > timeOutMs) + break; + } + } +} diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index 6e54d7402..599bf376c 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -16,7 +16,9 @@ package net.openhft.chronicle.map; +import com.google.common.collect.MapDifference; import com.google.common.collect.Maps; +import net.openhft.chronicle.hash.replication.ReplicableEntry; import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.lang.io.ByteBufferBytes; @@ -38,7 +40,6 @@ import java.util.concurrent.TimeUnit; */ public class TcpReplicationSoakTest { - private ChronicleMap map1; private ChronicleMap map2; static int s_port = 8093; @@ -50,6 +51,12 @@ public class TcpReplicationSoakTest { final InetSocketAddress endpoint = new InetSocketAddress("localhost", s_port + 1); + ChronicleMapBuilder builder = ChronicleMapBuilder + .of(Integer.class, CharSequence.class) + .entries(Builder.SIZE + Builder.SIZE) + .actualSegments(1) + .averageValue("test" + 1000); + { final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port, endpoint).autoReconnectedUponDroppedConnection(true) @@ -57,16 +64,12 @@ public class TcpReplicationSoakTest { .tcpBufferSize(1024 * 64); - map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(Builder.SIZE + Builder.SIZE) - .removedEntryCleanupTimeout(1, TimeUnit.MILLISECONDS) - .actualSegments(1) - .averageValue("test" + 1000) - .replication(SingleChronicleHashReplication.builder() + map1 = builder + .instance() + .replicated(SingleChronicleHashReplication.builder() .tcpTransportAndNetwork(tcpConfig1) .name("map1") .createWithId((byte) 1)) - .instance() .name("map1") .create(); } @@ -76,18 +79,14 @@ public class TcpReplicationSoakTest { .heartBeatInterval(1, TimeUnit.SECONDS) .tcpBufferSize(1024 * 64); - map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(Builder.SIZE + Builder.SIZE) - .removedEntryCleanupTimeout(1, TimeUnit.MILLISECONDS) - .averageValue("test" + 1000) - .replication(SingleChronicleHashReplication.builder() + map2 = builder + .instance() + .replicated(SingleChronicleHashReplication.builder() .tcpTransportAndNetwork(tcpConfig2) .name("map2") .createWithId((byte) 2)) - .instance() .name("map2") .create(); - } s_port += 2; } @@ -144,7 +143,24 @@ public class TcpReplicationSoakTest { waitTillEqual(15000); - Assert.assertEquals(map1, map2); + if (map1.equals(map2)) { + System.out.println("same"); + return; + } + MapDifference difference = Maps.difference(map1, map2); + Map onlyOnMap1 = difference.entriesOnlyOnLeft(); + System.out.println("only on map1:"); + for (Integer k : onlyOnMap1.keySet()) { + printState(map1, k); + } + Map onlyOnMap2 = difference.entriesOnlyOnRight(); + System.out.println("only on map2:"); + for (Integer k : onlyOnMap2.keySet()) { + printState(map2, k); + } + if (!map1.equals(map2)) { + Assert.assertEquals(difference.toString(), map1, map2); + } } finally { map1.close(); map2.close(); @@ -152,31 +168,13 @@ public class TcpReplicationSoakTest { } - @Test - public void testFloodReplicatedMapWithDeletedEntries() throws InterruptedException { - try { - Random r = ThreadLocalRandom.current(); - NavigableSet put = new TreeSet<>(); - for (int i = 0; i < Builder.SIZE * 10; i++) { - if (r.nextBoolean() || put.isEmpty()) { - int key = r.nextInt(); - map1.put(key, "test"); - put.add(key); - } else { - Integer key = put.pollFirst(); - map1.remove(key); - } -// Thread.sleep(0, 1000); - } - System.out.println("\nwaiting till equal"); - - waitTillEqual(15000); - - if (!map1.equals(map2)) - Assert.assertEquals(Maps.difference(map1, map2).toString(), map1, map2); - } finally { - map1.close(); - map2.close(); + private void printState(ChronicleMap m, Integer k) { + try (ExternalMapQueryContext q = m.queryContext(k)) { + MapEntry entry = q.entry(); + ReplicableEntry re = (ReplicableEntry) entry; + System.out.println(k + "=" + entry.value() + ", isChanged=" + + re.isChanged() + " oTs: " + re.originTimestamp() + + " oId: " + re.originIdentifier()); } }