From e0da2a2e9c84f5bd7766d1c7c20c7d2559e10e53 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 19 Aug 2014 19:23:38 +0100 Subject: [PATCH] HCOLL-125 Issue with cluster replication where an unwanted sleep(100) had to be include --- pom.xml | 14 ++ .../chronicle/map/ChronicleMapBuilder.java | 2 +- .../collections/ClusterReplicator.java | 21 ++- .../collections/ClusterReplicatorBuilder.java | 9 +- .../collections/SharedHashMapBuilder.java | 2 +- .../collections/ClusterReplicationTest.java | 20 +- .../PostClusterMapCreationTest.java | 170 +++++++++++++++++ .../PreClusterMapCreationTest.java | 173 ++++++++++++++++++ .../collections/SingleMapClusterTest.java | 166 +++++++++++++++++ 9 files changed, 560 insertions(+), 17 deletions(-) create mode 100644 src/test/java/net/openhft/collections/PostClusterMapCreationTest.java create mode 100644 src/test/java/net/openhft/collections/PreClusterMapCreationTest.java create mode 100644 src/test/java/net/openhft/collections/SingleMapClusterTest.java diff --git a/pom.xml b/pom.xml index d4a0cf70d..19ce22eca 100755 --- a/pom.xml +++ b/pom.xml @@ -89,6 +89,20 @@ + + org.slf4j + slf4j-api + + + + + + org.slf4j + slf4j-simple + test + + + diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index cbd7d1af1..7b7361963 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -138,7 +138,7 @@ public class ChronicleMapBuilder extends SharedHashMapBuilder implem } - public SharedHashMapBuilder file(File file) { + public ChronicleMapBuilder file(File file) { super.file(file); return this; } diff --git a/src/main/java/net/openhft/collections/ClusterReplicator.java b/src/main/java/net/openhft/collections/ClusterReplicator.java index 17fc9b78c..aa90707e3 100644 --- a/src/main/java/net/openhft/collections/ClusterReplicator.java +++ b/src/main/java/net/openhft/collections/ClusterReplicator.java @@ -34,6 +34,7 @@ import java.util.Queue; import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.LinkedTransferQueue; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReferenceArray; import static java.lang.Math.min; @@ -137,7 +138,7 @@ class ClusterReplicator implements ReplicaExternalizable, Closeable * @return a new DirectBitSet backed by a byteBuffer */ private static DirectBitSet newBitSet(int numberOfBits) { - final ByteBufferBytes bytes = new ByteBufferBytes(wrap(new byte[(numberOfBits+7) / 8])); + final ByteBufferBytes bytes = new ByteBufferBytes(wrap(new byte[(numberOfBits + 7) / 8])); return new SingleThreadedDirectBitSet(bytes); } @@ -149,9 +150,13 @@ class ClusterReplicator implements ReplicaExternalizable, Closeable chronicalChannels[chronicleChannel] = replica; chronicalChannelBitSet.set(chronicleChannel); + + // zero is used for bootstrapping if (chronicleChannel == 0) return; + + // boot strap messages are only sent if the cluster is already established for (int i = (int) systemModificationIteratorBitSet.nextSetBit(0); i > 0; i = (int) systemModificationIteratorBitSet.nextSetBit(i + 1)) { byte remoteIdentifier = (byte) i; final long lastModificationTime = replica.lastModificationTime(remoteIdentifier); @@ -227,9 +232,10 @@ class ClusterReplicator implements ReplicaExternalizable, Closeable public void dirtyEntries(long fromTimeStamp) { for (int i = (int) chronicalChannelBitSet.nextSetBit(0); i >= 0; i = (int) chronicalChannelBitSet.nextSetBit(i + 1)) { - chronicalChannels[i].acquireModificationIterator(remoteIdentifier, notifier).dirtyEntries(fromTimeStamp); - notifier.onChange(); + chronicalChannels[i].acquireModificationIterator(remoteIdentifier, + notifier).dirtyEntries(fromTimeStamp); } + notifier.onChange(); } }; @@ -247,7 +253,14 @@ class ClusterReplicator implements ReplicaExternalizable, Closeable */ @Override public long lastModificationTime(byte remoteIdentifier) { - long t = System.currentTimeMillis(); + + // if isEmpty, then we will ask for all the data from the remote node, as there could be a race + // around the chronicalChannelBitSet and the bootstrap message + if (chronicalChannelBitSet.nextSetBit(0) == -1) { + return 0; + } + + long t = System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(1); for (int i = (int) chronicalChannelBitSet.nextSetBit(0); i > 0; i = (int) chronicalChannelBitSet.nextSetBit(i + 1)) { t = min(t, chronicalChannels[i].lastModificationTime(remoteIdentifier)); } diff --git a/src/main/java/net/openhft/collections/ClusterReplicatorBuilder.java b/src/main/java/net/openhft/collections/ClusterReplicatorBuilder.java index 88382b787..278f25a8d 100644 --- a/src/main/java/net/openhft/collections/ClusterReplicatorBuilder.java +++ b/src/main/java/net/openhft/collections/ClusterReplicatorBuilder.java @@ -18,6 +18,8 @@ package net.openhft.collections; +import net.openhft.chronicle.map.ChronicleMapBuilder; + import java.io.Closeable; import java.io.IOException; import java.net.InetAddress; @@ -66,12 +68,13 @@ public class ClusterReplicatorBuilder { } /** - * @param chronicleChannel when clustering with a number of maps, each map will be called a chronicle channel + * @param chronicleChannel when clustering with a number of maps, each map will be called a chronicle + * channel */ - public SharedHashMap create(short chronicleChannel, SharedHashMapBuilder builder) throws + public SharedHashMap create(short chronicleChannel, ChronicleMapBuilder builder) throws IOException { - final SharedHashMapBuilder builder0 = builder.toBuilder(); + final ChronicleMapBuilder builder0 = builder.clone(); builder0.identifier(identifier); final VanillaSharedReplicatedHashMap result = diff --git a/src/main/java/net/openhft/collections/SharedHashMapBuilder.java b/src/main/java/net/openhft/collections/SharedHashMapBuilder.java index 049e777b4..01d26cb0a 100644 --- a/src/main/java/net/openhft/collections/SharedHashMapBuilder.java +++ b/src/main/java/net/openhft/collections/SharedHashMapBuilder.java @@ -327,7 +327,7 @@ public class SharedHashMapBuilder implements Cloneable { return file(file).kClass(kClass).vClass(vClass).create(); } - SharedHashMapBuilder toBuilder() throws IOException { + public SharedHashMapBuilder toBuilder() throws IOException { SharedHashMapBuilder builder = clone(); if (file != null) { diff --git a/src/test/java/net/openhft/collections/ClusterReplicationTest.java b/src/test/java/net/openhft/collections/ClusterReplicationTest.java index 8dab0a501..a5c3ce5b8 100644 --- a/src/test/java/net/openhft/collections/ClusterReplicationTest.java +++ b/src/test/java/net/openhft/collections/ClusterReplicationTest.java @@ -19,7 +19,11 @@ package net.openhft.collections; -import org.junit.*; +import net.openhft.chronicle.map.ChronicleMapBuilder; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; import java.io.Closeable; import java.io.File; @@ -71,7 +75,7 @@ public class ClusterReplicationTest { // this is how you add maps after the custer is created map1a = clusterReplicatorBuilder.create((short) 1, - SharedHashMapBuilder.of(Integer.class, CharSequence.class) + ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) .file(getPersistenceFile())); @@ -92,14 +96,14 @@ public class ClusterReplicationTest { // this is how you add maps after the custer is created map1b = clusterReplicatorBuilder1.create((short) 1, - SharedHashMapBuilder.of(Integer.class, CharSequence.class) + ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) .file(getPersistenceFile())); clusterB = clusterReplicatorBuilder1.create(); - } + } @After @@ -119,16 +123,16 @@ public class ClusterReplicationTest { public void test() throws IOException, InterruptedException { // todo remove this sleep - Thread.sleep(100); + + // Thread.sleep(100); map2b = clusterReplicatorBuilder1.create((short) 2, - SharedHashMapBuilder.of(Integer.class, CharSequence.class) + ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) .file(getPersistenceFile())); - map2a = clusterReplicatorBuilder.create((short) 2, - SharedHashMapBuilder.of(Integer.class, CharSequence.class) + ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) .file(getPersistenceFile())); diff --git a/src/test/java/net/openhft/collections/PostClusterMapCreationTest.java b/src/test/java/net/openhft/collections/PostClusterMapCreationTest.java new file mode 100644 index 000000000..de808410c --- /dev/null +++ b/src/test/java/net/openhft/collections/PostClusterMapCreationTest.java @@ -0,0 +1,170 @@ +/* + * Copyright 2014 Higher Frequency Trading + *

+ * http://www.higherfrequencytrading.com + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.collections; + + +import net.openhft.chronicle.map.ChronicleMapBuilder; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.Closeable; +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertTrue; + +/** + * Test VanillaSharedReplicatedHashMap where the Replicated is over a TCP Socket, but with 4 nodes + * + * @author Rob Austin. + */ +public class PostClusterMapCreationTest { + + private SharedHashMap map1a; + private SharedHashMap map2a; + + private SharedHashMap map1b; + private SharedHashMap map2b; + + private ClusterReplicator clusterB; + private ClusterReplicator clusterA; + + private ClusterReplicatorBuilder clusterReplicatorBuilder; + private ClusterReplicatorBuilder clusterReplicatorBuilder1; + + + public static File getPersistenceFile() { + String TMP = System.getProperty("java.io.tmpdir"); + File file = new File(TMP + "/test" + System.nanoTime()); + file.deleteOnExit(); + return file; + } + + + @Before + + public void setup() throws IOException { + + { + final TcpReplicatorBuilder tcpReplicatorBuilder = + new TcpReplicatorBuilder(8086, new InetSocketAddress("localhost", 8087)).heartBeatInterval(1, SECONDS); + + clusterReplicatorBuilder = new ClusterReplicatorBuilder((byte) 1, 1024); + clusterReplicatorBuilder.tcpReplicatorBuilder(tcpReplicatorBuilder); + clusterA = clusterReplicatorBuilder.create(); + + // this is how you add maps after the custer is created + map1a = clusterReplicatorBuilder.create((short) 1, + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(1000) + .file(getPersistenceFile())); + + map2a = clusterReplicatorBuilder.create((short) 2, + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(1000) + .file(getPersistenceFile())); + } + + + { + final TcpReplicatorBuilder tcpReplicatorBuilder = + new TcpReplicatorBuilder(8087).heartBeatInterval(1, SECONDS); + + clusterReplicatorBuilder1 = new ClusterReplicatorBuilder((byte) + 2, 1024); + clusterReplicatorBuilder1.tcpReplicatorBuilder(tcpReplicatorBuilder); + clusterB = clusterReplicatorBuilder1.create(); + + // this is how you add maps after the custer is created + map1b = clusterReplicatorBuilder1.create((short) 1, + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(1000) + .file(getPersistenceFile())); + + map2b = clusterReplicatorBuilder1.create((short) 2, + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(1000) + .file(getPersistenceFile())); + } + + + } + + @After + public void tearDown() throws InterruptedException { + + for (final Closeable closeable : new Closeable[]{clusterA, clusterB}) { + try { + closeable.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + } + + + /** + * sets up a 2 clusters ( notionally one for each host ), and replicates 2 Chronicle Maps between them + * + * @throws IOException + * @throws InterruptedException + */ + @Test + public void test() throws IOException, InterruptedException { + + map2a.put(1, "EXAMPLE-2"); + map1a.put(1, "EXAMPLE-1"); + + // allow time for the recompilation to resolve + waitTillEqual(2500); + + Assert.assertEquals("map1a=map1b", map1a, map1b); + Assert.assertEquals("map2a=map2b", map2a, map2b); + + assertTrue("map1a.empty", !map1a.isEmpty()); + assertTrue("map2a.empty", !map2a.isEmpty()); + + } + + + /** + * waits until map1 and map2 show the same value + * + * @param timeOutMs timeout in milliseconds + * @throws InterruptedException + */ + + private void waitTillEqual(final int timeOutMs) throws InterruptedException { + for (int t = 0; t < timeOutMs; t++) { + if (map1a.equals(map1b) && + map2a.equals(map2b)) + break; + Thread.sleep(1); + } + + } + +} + + + diff --git a/src/test/java/net/openhft/collections/PreClusterMapCreationTest.java b/src/test/java/net/openhft/collections/PreClusterMapCreationTest.java new file mode 100644 index 000000000..235d6143b --- /dev/null +++ b/src/test/java/net/openhft/collections/PreClusterMapCreationTest.java @@ -0,0 +1,173 @@ +/* + * Copyright 2014 Higher Frequency Trading + *

+ * http://www.higherfrequencytrading.com + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.collections; + + +import net.openhft.chronicle.map.ChronicleMapBuilder; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.Closeable; +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertTrue; + +/** + * Test VanillaSharedReplicatedHashMap where the Replicated is over a TCP Socket, but with 4 nodes + * + * @author Rob Austin. + */ +public class PreClusterMapCreationTest { + + private SharedHashMap map1a; + private SharedHashMap map2a; + + private SharedHashMap map1b; + private SharedHashMap map2b; + + private ClusterReplicator clusterB; + private ClusterReplicator clusterA; + + private ClusterReplicatorBuilder clusterReplicatorBuilder; + private ClusterReplicatorBuilder clusterReplicatorBuilder1; + + + public static File getPersistenceFile() { + String TMP = System.getProperty("java.io.tmpdir"); + File file = new File(TMP + "/test" + System.nanoTime()); + file.deleteOnExit(); + return file; + } + + + @Before + + public void setup() throws IOException { + + { + final TcpReplicatorBuilder tcpReplicatorBuilder = + new TcpReplicatorBuilder(8086, new InetSocketAddress("localhost", 8087)).heartBeatInterval(1, SECONDS); + + clusterReplicatorBuilder = new ClusterReplicatorBuilder((byte) 1, 1024); + clusterReplicatorBuilder.tcpReplicatorBuilder(tcpReplicatorBuilder); + + + // this is how you add maps after the custer is created + map1a = clusterReplicatorBuilder.create((short) 1, + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(1000) + .file(getPersistenceFile())); + + map2a = clusterReplicatorBuilder.create((short) 2, + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(1000) + .file(getPersistenceFile())); + + + clusterA = clusterReplicatorBuilder.create(); + + } + + + { + + final TcpReplicatorBuilder tcpReplicatorBuilder = + new TcpReplicatorBuilder(8087).heartBeatInterval(1, SECONDS); + + clusterReplicatorBuilder1 = new ClusterReplicatorBuilder((byte) + 2, 1024); + clusterReplicatorBuilder1.tcpReplicatorBuilder(tcpReplicatorBuilder); + + + // this is how you add maps after the custer is created + map1b = clusterReplicatorBuilder1.create((short) 1, + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(1000) + .file(getPersistenceFile())); + + map2b = clusterReplicatorBuilder1.create((short) 2, + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(1000) + .file(getPersistenceFile())); + + + clusterB = clusterReplicatorBuilder1.create(); + } + + + } + + @After + public void tearDown() throws InterruptedException { + + for (final Closeable closeable : new Closeable[]{clusterA, clusterB}) { + try { + closeable.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + } + + + @Test + public void test() throws IOException, InterruptedException { + + + map2a.put(1, "EXAMPLE-2"); + map1a.put(1, "EXAMPLE-1"); + + // allow time for the recompilation to resolve + waitTillEqual(2500); + + Assert.assertEquals("map1a=map1b", map1a, map1b); + Assert.assertEquals("map2a=map2b", map2a, map2b); + + assertTrue("map1a.empty", !map1a.isEmpty()); + assertTrue("map2a.empty", !map2a.isEmpty()); + + } + + + /** + * waits until map1 and map2 show the same value + * + * @param timeOutMs timeout in milliseconds + * @throws InterruptedException + */ + + private void waitTillEqual(final int timeOutMs) throws InterruptedException { + for (int t = 0; t < timeOutMs; t++) { + if (map1a.equals(map1b) && + map2a.equals(map2b)) + break; + Thread.sleep(1); + } + + } + +} + + + diff --git a/src/test/java/net/openhft/collections/SingleMapClusterTest.java b/src/test/java/net/openhft/collections/SingleMapClusterTest.java new file mode 100644 index 000000000..5e9f319ab --- /dev/null +++ b/src/test/java/net/openhft/collections/SingleMapClusterTest.java @@ -0,0 +1,166 @@ +/* + * Copyright 2014 Higher Frequency Trading + *

+ * http://www.higherfrequencytrading.com + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.collections; + + +import net.openhft.chronicle.map.ChronicleMapBuilder; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.Closeable; +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertTrue; + +/** + * Test VanillaSharedReplicatedHashMap where the Replicated is over a TCP Socket, but with 4 nodes + * + * @author Rob Austin. + */ +public class SingleMapClusterTest { + + private SharedHashMap map1a; + + private SharedHashMap map1b; + + private ClusterReplicator clusterB; + private ClusterReplicator clusterA; + + private ClusterReplicatorBuilder clusterReplicatorBuilder; + private ClusterReplicatorBuilder clusterReplicatorBuilder1; + + + public static File getPersistenceFile() { + String TMP = System.getProperty("java.io.tmpdir"); + File file = new File(TMP + "/test" + System.nanoTime()); + file.deleteOnExit(); + return file; + } + + + @Before + + public void setup() throws IOException { + + { + final TcpReplicatorBuilder tcpReplicatorBuilder = + new TcpReplicatorBuilder(8086, new InetSocketAddress("localhost", 8087)).heartBeatInterval(1, SECONDS); + + clusterReplicatorBuilder = new ClusterReplicatorBuilder((byte) 1, 1024); + clusterReplicatorBuilder.tcpReplicatorBuilder(tcpReplicatorBuilder); + // this is how you add maps after the custer is created + map1a = clusterReplicatorBuilder.create((short) 1, + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(1000) + .file(getPersistenceFile())); + + clusterA = clusterReplicatorBuilder.create(); + + + + + + + + } + + + { + + final TcpReplicatorBuilder tcpReplicatorBuilder = + new TcpReplicatorBuilder(8087).heartBeatInterval(1, SECONDS); + + clusterReplicatorBuilder1 = new ClusterReplicatorBuilder((byte) + 2, 1024); + clusterReplicatorBuilder1.tcpReplicatorBuilder(tcpReplicatorBuilder); + + + // this is how you add maps after the custer is created + map1b = clusterReplicatorBuilder1.create((short) 1, + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(1000) + .file(getPersistenceFile())); + + clusterB = clusterReplicatorBuilder1.create(); + + } + + + } + + @After + public void tearDown() throws InterruptedException { + + for (final Closeable closeable : new Closeable[]{clusterA, clusterB}) { + try { + closeable.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + } + + + @Test + public void test() throws IOException, InterruptedException { + + // todo remove this sleep + + + + map1a.put(1, "EXAMPLE-1"); + + // allow time for the recompilation to resolve + waitTillEqual(2500); + + Assert.assertEquals("map1a=map1b", map1a, map1b); + + + assertTrue("map1a.empty", !map1a.isEmpty()); + + + } + + + /** + * waits until map1 and map2 show the same value + * + * @param timeOutMs timeout in milliseconds + * @throws InterruptedException + */ + + private void waitTillEqual(final int timeOutMs) throws InterruptedException { + for (int t = 0; t < timeOutMs; t++) { + if (map1a.equals(map1b) ) + + break; + Thread.sleep(1); + } + + } + +} + + +