From c4417ddbe49e0fcc8e6ccaa6456265546a39662b Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 2 Dec 2014 20:08:26 +0000 Subject: [PATCH] adding net.openhft.chronicle.map.NoTcpReplicationSoakTest --- .../chronicle/map/ReplicatedChronicleMap.java | 150 +++++++----- .../openhft/chronicle/map/TcpReplicator.java | 19 +- .../map/NoTcpReplicationSoakTest.java | 230 ++++++++++++++++++ .../map/TCPSocketReplicationTest.java | 45 +++- .../chronicle/map/TcpReplicationSoakTest.java | 64 +++-- .../map/TcpTimeBasedReplicationSoakTest.java | 166 +++++++++++++ 6 files changed, 579 insertions(+), 95 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java create mode 100644 src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 9b517eb0c..0ce3d5567 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -48,36 +48,39 @@ import static net.openhft.lang.MemoryUnit.*; import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND; /** - *

A Replicating Multi Master HashMap

Each remote hash map, mirrors its changes over to another - * remote hash map, neither hash map is considered the master store of data, each hash map uses timestamps to - * reconcile changes. We refer to an instance of a remote hash-map as a node. A node will be connected to any - * number of other nodes, for the first implementation the maximum number of nodes will be fixed. The data - * that is stored locally in each node will become eventually consistent. So changes made to one node, for - * example by calling put() will be replicated over to the other node. To achieve a high level of performance - * and throughput, the call to put() won’t block, with concurrentHashMap, It is typical to check the return - * code of some methods to obtain the old value for example remove(). Due to the loose coupling and lock free - * nature of this multi master implementation, this return value will only be the old value on the nodes - * local data store. In other words the nodes are only concurrent locally. Its worth realising that another - * node performing exactly the same operation may return a different value. However reconciliation will ensure - * the maps themselves become eventually consistent.

Reconciliation

If two ( or more nodes ) - * were to receive a change to their maps for the same key but different values, say by a user of the maps, - * calling the put(key, value). Then, initially each node will update its local store and each local store - * will hold a different value, but the aim of multi master replication is to provide eventual consistency - * across the nodes. So, with multi master when ever a node is changed it will notify the other nodes of its - * change. We will refer to this notification as an event. The event will hold a timestamp indicating the time - * the change occurred, it will also hold the state transition, in this case it was a put with a key and - * value. Eventual consistency is achieved by looking at the timestamp from the remote node, if for a given - * key, the remote nodes timestamp is newer than the local nodes timestamp, then the event from the remote - * node will be applied to the local node, otherwise the event will be ignored.

However there is an - * edge case that we have to concern ourselves with, If two nodes update their map at the same time with - * different values, we have to deterministically resolve which update wins, because of eventual consistency - * both nodes should end up locally holding the same data. Although it is rare two remote nodes could receive - * an update to their maps at exactly the same time for the same key, we have to handle this edge case, its - * therefore important not to rely on timestamps alone to reconcile the updates. Typically the update with the - * newest timestamp should win, but in this example both timestamps are the same, and the decision made to one - * node should be identical to the decision made to the other. We resolve this simple dilemma by using a node - * identifier, each node will have a unique identifier, the update from the node with the smallest identifier - * wins.

+ *

A Replicating Multi Master HashMap

Each remote hash map, mirrors its changes over to + * another remote hash map, neither hash map is considered the master store of data, each hash map + * uses timestamps to reconcile changes. We refer to an instance of a remote hash-map as a node. A + * node will be connected to any number of other nodes, for the first implementation the maximum + * number of nodes will be fixed. The data that is stored locally in each node will become + * eventually consistent. So changes made to one node, for example by calling put() will be + * replicated over to the other node. To achieve a high level of performance and throughput, the + * call to put() won’t block, with concurrentHashMap, It is typical to check the return code of some + * methods to obtain the old value for example remove(). Due to the loose coupling and lock free + * nature of this multi master implementation, this return value will only be the old value on the + * nodes local data store. In other words the nodes are only concurrent locally. Its worth realising + * that another node performing exactly the same operation may return a different value. However + * reconciliation will ensure the maps themselves become eventually consistent.

+ *

Reconciliation

If two ( or more nodes ) were to receive a change to their maps for + * the same key but different values, say by a user of the maps, calling the put(key, value). Then, + * initially each node will update its local store and each local store will hold a different value, + * but the aim of multi master replication is to provide eventual consistency across the nodes. So, + * with multi master when ever a node is changed it will notify the other nodes of its change. We + * will refer to this notification as an event. The event will hold a timestamp indicating the time + * the change occurred, it will also hold the state transition, in this case it was a put with a key + * and value. Eventual consistency is achieved by looking at the timestamp from the remote node, if + * for a given key, the remote nodes timestamp is newer than the local nodes timestamp, then the + * event from the remote node will be applied to the local node, otherwise the event will be + * ignored.

However there is an edge case that we have to concern ourselves with, If two + * nodes update their map at the same time with different values, we have to deterministically + * resolve which update wins, because of eventual consistency both nodes should end up locally + * holding the same data. Although it is rare two remote nodes could receive an update to their maps + * at exactly the same time for the same key, we have to handle this edge case, its therefore + * important not to rely on timestamps alone to reconcile the updates. Typically the update with the + * newest timestamp should win, but in this example both timestamps are the same, and the decision + * made to one node should be identical to the decision made to the other. We resolve this simple + * dilemma by using a node identifier, each node will have a unique identifier, the update from the + * node with the smallest identifier wins.

* * @param the entries key type * @param the entries value type @@ -207,7 +210,8 @@ final class ReplicatedChronicleMap>instance(), BytesBytesInterop.INSTANCE, key, keySize, keyBytesToInstance, getRemoteBytesValueInterops, value, valueBytesToInstance, - replaceIfPresent, readValue, false, localIdentifier, currentTime());; + replaceIfPresent, readValue, false, localIdentifier, currentTime()); + ; } /** @@ -239,8 +243,8 @@ final class ReplicatedChronicleMap, RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> @@ -380,8 +412,8 @@ final class ReplicatedChronicleMap, RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> @@ -1133,8 +1162,8 @@ final class ReplicatedChronicleMapOnce a change occurs to a map, map replication requires that these changes are picked up by another - * thread, this class provides an iterator like interface to poll for such changes.

In most cases - * the thread that adds data to the node is unlikely to be the same thread that replicates the data over - * to the other nodes, so data will have to be marshaled between the main thread storing data to the map, - * and the thread running the replication.

One way to perform this marshalling, would be to pipe - * the data into a queue. However, This class takes another approach. It uses a bit set, and marks bits - * which correspond to the indexes of the entries that have changed. It then provides an iterator like - * interface to poll for such changes.

+ *

Once a change occurs to a map, map replication requires that these changes are picked up + * by another thread, this class provides an iterator like interface to poll for such changes. + *

In most cases the thread that adds data to the node is unlikely to be the same thread + * that replicates the data over to the other nodes, so data will have to be marshaled between + * the main thread storing data to the map, and the thread running the replication.

One + * way to perform this marshalling, would be to pipe the data into a queue. However, This class + * takes another approach. It uses a bit set, and marks bits which correspond to the indexes of + * the entries that have changed. It then provides an iterator like interface to poll for such + * changes.

* * @author Rob Austin. */ @@ -1344,7 +1374,8 @@ final class ReplicatedChronicleMap 0) - contemplateThrottleWrites(bytesJustWritten); + if (len == -1) + socketChannel.close(); + + if (len > 0) + contemplateThrottleWrites(len); if (!attached.entryWriter.hasBytesToWrite() && !attached.entryWriter.isWorkIncomplete()) // TURN OP_WRITE_OFF @@ -659,6 +662,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable int len = attached.entryReader.readSocketToBuffer(socketChannel); if (len == -1) { + socketChannel.register(selector, 0); if (replicationConfig.autoReconnectedUponDroppedConnection()) { AbstractConnector connector = attached.connector; if (connector != null) @@ -1018,6 +1022,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // if we still have some unwritten writer from last time lastSentTime = approxTime; + assert in.position() <= Integer.MAX_VALUE; int size = (int) in.position(); out.limit(size); @@ -1027,7 +1032,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable LOG.debug("bytes-written=" + len); if (len == size) { - // assert len == size; out.clear(); in.clear(); } else { @@ -1195,9 +1199,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable final long limit = out.limit(); out.limit(nextEntryPos); - boolean isStateless = (state != 1); + boolean isStatelessClient = (state != 1); - if (isStateless) { + if (isStatelessClient) { if (statelessServerConnector == null) { LOG.error("", new IllegalArgumentException("received an event " + "from a stateless map, stateless maps are not " + @@ -1247,6 +1251,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable return; in.limit(in.position()); + assert out.position() < Integer.MAX_VALUE; in.position((int) out.position()); in.compact(); @@ -1268,8 +1273,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable return out.readLong(); else return Long.MIN_VALUE; - - // return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE; } public long remoteHeartbeatIntervalFromBuffer() { diff --git a/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java new file mode 100644 index 000000000..17a5c256b --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java @@ -0,0 +1,230 @@ +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.replication.TimeProvider; +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.*; + +import java.io.Closeable; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.*; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * @author Rob Austin. + */ +public class NoTcpReplicationSoakTest { + + private ReplicatedChronicleMap map1; + private ReplicatedChronicleMap map2; + private IntValue value; + static int s_port = 8010; + private long time; + + TimeProvider timeProvider = new TimeProvider() { + @Override + public long currentTimeMillis() { + return time; + } + }; + + + @Before + public void setup() throws IOException { + value = DataValueClasses.newDirectReference(IntValue.class); + ((Byteable) value).bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0); + + + { + + + map1 = (ReplicatedChronicleMap) ChronicleMapBuilder.of(Integer.class, CharSequence.class) + + .name("map1") + .replication((byte) 1) + .timeProvider(timeProvider) + .create(); + } + { + + map2 = (ReplicatedChronicleMap) ChronicleMapBuilder.of(Integer.class, CharSequence.class) + + .name("map2") + .replication((byte) 2) + .timeProvider(timeProvider) + .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); + } + + AtomicInteger task = new AtomicInteger(); + + @Test + @Ignore + public void testSoakTestWithRandomData() throws IOException, InterruptedException { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + System.out.print("SoakTesting "); + for (int j = 1; j < 10000; j++) { + if (j % 100 == 0) + System.out.print("."); + + Random rnd = new Random(j); + + + if (rnd.nextInt(10) < 2) + time++; + + final long t = time; + + final int key = rnd.nextInt(1000); + + // add random delay + addRandomDelay(executorService, rnd); + + if (rnd.nextBoolean()) { + final CharSequence value = "test" + j; + if (rnd.nextBoolean()) { + map1.put(key, value); + + task.incrementAndGet(); + executorService.submit(new Runnable() { + @Override + public void run() { + task.decrementAndGet(); + map2.put0(key, value, true, (byte) 1, t); + } + }); + + } else { + map2.put(key, value); + + task.incrementAndGet(); + executorService.submit(new Runnable() { + @Override + public void run() { + task.decrementAndGet(); + map1.put0(key, value, true, (byte) 2, t); + } + }); + } + + } else { + + if (rnd.nextBoolean()) { + map1.remove(key); + + task.incrementAndGet(); + executorService.submit(new Runnable() { + @Override + public void run() { + task.decrementAndGet(); + map2.remoteRemove(key, (byte) 1, time); + } + }); + + } else { + map2.remove(key); + + task.incrementAndGet(); + executorService.submit(new Runnable() { + @Override + public void run() { + task.decrementAndGet(); + map1.remoteRemove(key, (byte) 2, time); + } + }); + } + } + } + + while (task.get() > 0) { + + } + + + Assert.assertEquals(new TreeMap(map1), new TreeMap(map2)); + + } + + private void addRandomDelay(ExecutorService executorService, Random rnd) { + if (rnd.nextInt(30) < 2) { + + task.incrementAndGet(); + executorService.submit(new Runnable() { + @Override + public void run() { + task.decrementAndGet(); + try { + Thread.sleep(1); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + }); + } + } + + + 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(500); + if (numberOfTimesTheSame == 100) { + System.out.println("same"); + break; + } + + } + Thread.sleep(1); + } + } + +} + diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java index fff7574f5..32df4b0bd 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java @@ -28,7 +28,6 @@ import java.net.InetSocketAddress; import java.util.Random; import java.util.Set; -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; @@ -52,7 +51,7 @@ public class TCPSocketReplicationTest { ChronicleMapBuilder map1Builder = newTcpSocketShmBuilder(Integer.class, CharSequence.class, (byte) 1, port, new InetSocketAddress("localhost", port + 1)); - map1 = map1Builder.createPersistedTo(getPersistenceFile()); + map1 = map1Builder.create(); map2 = newTcpSocketShmIntString((byte) 2, port + 1); s_port += 2; } @@ -116,6 +115,46 @@ public class TCPSocketReplicationTest { assertTrue(!map1.isEmpty()); } + + @Test + public void test2() throws IOException, InterruptedException { + + map2.put(1, "EXAMPLE-1"); + map2.remove(1); + + map1.put(1, "EXAMPLE-2"); + map1.remove(1); + + map2.put(1, "EXAMPLE-1"); + + + // allow time for the recompilation to resolve + waitTillEqual(5000); + + assertEquals(map1, map2); + assertTrue(!map1.isEmpty()); + } + + @Test + public void test4() throws IOException, InterruptedException { + + map2.put(1, "EXAMPLE-1"); + map2.remove(1); + + map1.put(1, "EXAMPLE-2"); + map1.remove(1); + + map2.put(1, "EXAMPLE-1"); + map2.remove(1); + + // allow time for the recompilation to resolve + waitTillEqual(5000); + + assertEquals(map1, map2); + assertTrue(map1.isEmpty()); + } + + @Test public void testBufferOverflow() throws IOException, InterruptedException { @@ -168,7 +207,7 @@ public class TCPSocketReplicationTest { } } System.out.println(""); - waitTillEqual(10000); + waitTillEqual(1000); } } diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index bc538fdf5..825f7b2da 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -11,9 +11,7 @@ import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; -import java.util.Random; -import java.util.Set; -import java.util.concurrent.TimeUnit; +import java.util.*; /** * @author Rob Austin. @@ -35,25 +33,26 @@ public class TcpReplicationSoakTest { { final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port, - endpoint) - .heartBeatInterval(1L, TimeUnit.SECONDS) - .autoReconnectedUponDroppedConnection(true); + endpoint).autoReconnectedUponDroppedConnection(true).name(" map1"); map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(Builder.SIZE) + .entries(Builder.SIZE + Builder.SIZE) + .actualSegments(1) + .name("map1") .replication((byte) 1, tcpConfig1) .create(); } { - final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of(s_port + 1) - .heartBeatInterval(1L, TimeUnit.SECONDS) - .autoReconnectedUponDroppedConnection(true); + final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of + (s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2"); map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(Builder.SIZE) + .entries(Builder.SIZE + Builder.SIZE) + .name("map2") .replication((byte) 2, tcpConfig2) .create(); + } s_port += 2; } @@ -86,8 +85,8 @@ public class TcpReplicationSoakTest { } - // TODO test this with larger sizes. - @Test(timeout = 20000) + @Test + @Ignore public void testSoakTestWithRandomData() throws IOException, InterruptedException { System.out.print("SoakTesting "); @@ -99,32 +98,47 @@ public class TcpReplicationSoakTest { 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)); - } + 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); + waitTillEqual(10000); - Assert.assertEquals(map1, map2); + Assert.assertEquals(new TreeMap(map1), new TreeMap(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)) { - numberOfTimesTheSame++; - Thread.sleep(1); - if (numberOfTimesTheSame == 100) + if (map1.equals(map1UnChanged) && map2.equals(map2UnChanged)) { + numberOfTimesTheSame++; + } else { + numberOfTimesTheSame = 0; + map1UnChanged = new HashMap(map1); + map2UnChanged = new HashMap(map2); + } + Thread.sleep(500); + if (numberOfTimesTheSame == 100) { + System.out.println("same"); break; - } else numberOfTimesTheSame = 0; + } + + } Thread.sleep(1); } } } + diff --git a/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java new file mode 100644 index 000000000..d2e5b5720 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java @@ -0,0 +1,166 @@ +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.hash.replication.TimeProvider; +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.*; +import org.mockito.Mockito; + +import java.io.Closeable; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.util.*; +import java.util.concurrent.TimeUnit; + +/** + * @author Rob Austin. + */ +public class TcpTimeBasedReplicationSoakTest { + + + private ChronicleMap map1; + private ChronicleMap map2; + private IntValue value; + static int s_port = 8010; + private TimeProvider timeProvider; + int t = 0; + + @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); + timeProvider = new TimeProvider() { + + Random rnd = new Random(4); + + @Override + public long currentTimeMillis() { + + if (rnd.nextBoolean()) + return t++; + else + return t; + } + }; + + { + final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port, + endpoint); + + + map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(Builder.SIZE) + .timeProvider(timeProvider) + .replication((byte) 1, tcpConfig1) + .create(); + } + { + final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of(s_port + 1); + + map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(Builder.SIZE) + .timeProvider(timeProvider) + .replication((byte) 2, tcpConfig2) + .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 + @Ignore + public void testSoakTestWithRandomData() throws IOException, InterruptedException { + + Thread.sleep(100); + + // for (int i = 30; i < 50; i++) { + System.out.print("SoakTesting "); + int max = 1000000; + for (int j = 1; j < max; j++) { + if (j % 100 == 0) + System.out.print("."); + Random rnd = new Random(System.currentTimeMillis()); + + final ChronicleMap map = rnd.nextBoolean() ? map1 : map2; + + if (rnd.nextBoolean()) { + map.put((int) rnd.nextInt(100), "test" + j); + } else { + map.remove((int) rnd.nextInt(100)); + } + + } + + System.out.println("\nwaiting till equal"); + + waitTillUnchanged(1000); + System.out.println("time t=" + t); + Assert.assertEquals(new TreeMap(map1), new TreeMap(map2)); + + } + + + private void waitTillUnchanged(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(map1UnChanged) && map2.equals(map2UnChanged)) { + numberOfTimesTheSame++; + } else { + numberOfTimesTheSame = 0; + map1UnChanged = new HashMap(map1); + map2UnChanged = new HashMap(map2); + } + Thread.sleep(50); + if (numberOfTimesTheSame == 100) { + + break; + } + + + } + } + + private void late(TimeProvider timeProvider) { + Mockito.when(timeProvider.currentTimeMillis()).thenReturn(System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(5)); + } +} +