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