HCOLL-125 Issue with cluster replication where an unwanted sleep(100) had to be include

pom.xml

@@ -89,6 +89,20 @@
         </dependency>
 
 
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+
+        <!-- test dependencies -->
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+
     </dependencies>
 
 

src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java

@@ -138,7 +138,7 @@ public class ChronicleMapBuilder<K, V> extends SharedHashMapBuilder<K, V> implem
     }
 
 
-    public SharedHashMapBuilder<K, V> file(File file) {
+    public ChronicleMapBuilder<K, V> file(File file) {
         super.file(file);
         return this;
     }

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<K, V> implements ReplicaExternalizable<K, V>, 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<K, V> implements ReplicaExternalizable<K, V>, 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<K, V> implements ReplicaExternalizable<K, V>, 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<K, V> implements ReplicaExternalizable<K, V>, 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));
         }

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 <K, V> SharedHashMap<K, V> create(short chronicleChannel, SharedHashMapBuilder<K, V> builder) throws
+    public <K, V> SharedHashMap<K, V> create(short chronicleChannel, ChronicleMapBuilder<K, V> builder) throws
             IOException {
 
-        final SharedHashMapBuilder<K, V> builder0 = builder.toBuilder();
+        final ChronicleMapBuilder<K, V> builder0 = builder.clone();
         builder0.identifier(identifier);
 
         final VanillaSharedReplicatedHashMap<K, V> result =

src/main/java/net/openhft/collections/SharedHashMapBuilder.java

@@ -327,7 +327,7 @@ public class SharedHashMapBuilder<K, V> implements Cloneable {
         return file(file).kClass(kClass).vClass(vClass).create();
     }
 
-    SharedHashMapBuilder<K, V> toBuilder() throws IOException {
+    public SharedHashMapBuilder<K, V> toBuilder() throws IOException {
         SharedHashMapBuilder builder = clone();
 
         if (file != null) {

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

src/test/java/net/openhft/collections/PostClusterMapCreationTest.java

@@ -0,0 +1,170 @@
+/*
+ * Copyright 2014 Higher Frequency Trading
+ * <p/>
+ * http://www.higherfrequencytrading.com
+ * <p/>
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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<Integer, CharSequence> map1a;
+    private SharedHashMap<Integer, CharSequence> map2a;
+
+    private SharedHashMap<Integer, CharSequence> map1b;
+    private SharedHashMap<Integer, CharSequence> 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);
+        }
+
+    }
+
+}
+
+
+

src/test/java/net/openhft/collections/PreClusterMapCreationTest.java

@@ -0,0 +1,173 @@
+/*
+ * Copyright 2014 Higher Frequency Trading
+ * <p/>
+ * http://www.higherfrequencytrading.com
+ * <p/>
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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<Integer, CharSequence> map1a;
+    private SharedHashMap<Integer, CharSequence> map2a;
+
+    private SharedHashMap<Integer, CharSequence> map1b;
+    private SharedHashMap<Integer, CharSequence> 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);
+        }
+
+    }
+
+}
+
+
+

src/test/java/net/openhft/collections/SingleMapClusterTest.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright 2014 Higher Frequency Trading
+ * <p/>
+ * http://www.higherfrequencytrading.com
+ * <p/>
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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<Integer, CharSequence> map1a;
+
+    private SharedHashMap<Integer, CharSequence> 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);
+        }
+
+    }
+
+}
+
+
+