Merge remote-tracking branch 'origin/master'

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

@@ -166,7 +166,10 @@ public class EngineMap<K, V> implements ChronicleMap<K, V> {
     @Override
     public V put(final K key, final V value) {
         nullCheck(key);
-        return toObject(vClass, () -> map.put(bytes(key), bytes(value)));
+        return toObject(vClass, () -> {
+            final byte[] put = map.put(bytes(key), bytes(value));
+            return put;
+        });
     }
 
     @Override

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

@@ -13,17 +13,35 @@ import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 /**
- *
+ * A {@link Map} implementation that stores each entry as a file in a
+ * directory. The <code>key</code> is the file name and the <code>value</code>
+ * is the contents of the file. This map will only handle <code>String</code>'s.
+ * <p>
+ * The class is effectively an abstraction over a directory in the file system.
+ * Therefore when the underlying files are changed an event will be fired to those
+ * registered for notifications.
+ * Since every write to this map will cause a change to underlying
+ * file system the event will distinguish between a programmatic event
+ * (i.e one caused my the actions of the map itself) and an event that
+ * has been triggered as a direct result of a file being manipulated outside
+ * this class.
+ * <p>
+ * Updates will be fired every time the file is saved but will be suppressed
+ * if the value has not changed.  To avoid temporary files (e.g. if edited in vi)
+ * being included in the map, any file starting with a '.' will be ignored.
+ * <p>
+ * Note the {@link WatchService} is extremely OS dependant.  Mas OSX registers
+ * very few events if they are done quickly and there is a significant delay
+ * between the event and the event being triggered.
  */
 public class FilePerKeyMap implements Map<String, String> {
-    private final String dir;
     private final Path dirPath;
     private final Map<File, Long> lastModifiedByProgram = new ConcurrentHashMap<>();
+    private final Map<File, String> lastUpdate = new ConcurrentHashMap<>();
     private final List<Consumer<FPMEvent>> listeners = new ArrayList<>();
     private final FPMWatcher fileFpmWatcher = new FPMWatcher();
 
     public FilePerKeyMap(String dir){
-        this.dir = dir;
         this.dirPath = Paths.get(dir);
         try {
             Files.createDirectories(dirPath);
@@ -59,7 +77,6 @@ public class FilePerKeyMap implements Map<String, String> {
 
     @Override
     public boolean containsKey(Object key) {
-
         return getFiles().anyMatch(p->p.getFileName().toString().equals(key));
     }
 
@@ -87,10 +104,9 @@ public class FilePerKeyMap implements Map<String, String> {
     public String remove(Object key) {
         String existing = get(key);
         if(existing != null){
-            deleteFile(Paths.get(dir, (String)key));
+            deleteFile(dirPath.resolve((String) key));
         }
         return existing;
-
     }
 
     @Override
@@ -132,7 +148,7 @@ public class FilePerKeyMap implements Map<String, String> {
         try {
             return Files.walk(dirPath).filter(p -> !Files.isDirectory(p));
         } catch (IOException e) {
-            throw new AssertionError(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -140,9 +156,9 @@ public class FilePerKeyMap implements Map<String, String> {
         String existingValue = null;
         if(Files.exists(path)) {
             try {
-                existingValue = Files.readAllLines(path).stream().collect(Collectors.joining());
+                existingValue = Files.readAllLines(path).stream().collect(Collectors.joining("\n"));
             } catch (IOException e) {
-                throw new AssertionError(e);
+                throw new RuntimeException(e);
             }
         }
         return existingValue;
@@ -208,7 +224,6 @@ public class FilePerKeyMap implements Map<String, String> {
 
             if (key != null ? !key.equals(fpmEntry.key) : fpmEntry.key != null) return false;
             return !(value != null ? !value.equals(fpmEntry.value) : fpmEntry.value != null);
-
         }
 
         @Override
@@ -236,21 +251,24 @@ public class FilePerKeyMap implements Map<String, String> {
                     try {
                         key = watcher.take();
                         for (WatchEvent<?> event : key.pollEvents()) {
-                            // get event type
                             WatchEvent.Kind<?> kind = event.kind();
 
                             // get file name
                             WatchEvent<Path> ev = (WatchEvent<Path>) event;
                             Path fileName = ev.context();
-
                             String mapKey = fileName.toString();
 
+                            if(mapKey.startsWith(".")){
+                                //this avoids temporary files being added to the map
+                                continue;
+                            }
 
                             if (kind == StandardWatchEventKinds.OVERFLOW) {
                                 continue;
                             } else if (kind == StandardWatchEventKinds.ENTRY_CREATE) {
                                 Path p = dirPath.resolve(fileName);
                                 String mapVal = getFileContents(p);
+                                lastUpdate.put(p.toFile(), mapVal);
                                 if(isProgrammaticUpdate(p.toFile())){
                                     fireEvent(new FPMEvent(FPMEvent.EventType.NEW, true, mapKey,mapVal));
                                 }else {
@@ -258,15 +276,25 @@ public class FilePerKeyMap implements Map<String, String> {
                                 }
                             } else if (kind == StandardWatchEventKinds.ENTRY_DELETE) {
                                 Path p = dirPath.resolve(fileName);
+                                lastUpdate.remove(p.toFile());
                                 if(isProgrammaticUpdate(p.toFile())){
                                     fireEvent(new FPMEvent(FPMEvent.EventType.DELETE, true, mapKey, null));
-                                    lastModifiedByProgram.remove(p.toFile());
                                 }else {
                                     fireEvent(new FPMEvent(FPMEvent.EventType.DELETE, false, mapKey, null));
                                 }
                             } else if (kind == StandardWatchEventKinds.ENTRY_MODIFY) {
                                 Path p = dirPath.resolve(fileName);
                                 String mapVal = getFileContents(p);
+                                String lastVal = null;
+                                if(mapVal != null) {
+                                    lastVal = lastUpdate.put(p.toFile(), mapVal);
+                                }
+
+                                if(lastVal != null && lastVal.equals(mapVal)){
+                                    //Nothing has changed don't fire an event
+                                    continue;
+                                }
+
                                 if(isProgrammaticUpdate(p.toFile())){
                                     fireEvent(new FPMEvent(FPMEvent.EventType.UPDATE, true, mapKey,mapVal));
                                 }else {
@@ -281,7 +309,7 @@ public class FilePerKeyMap implements Map<String, String> {
                     }
                 }
             } catch (IOException e) {
-                throw new AssertionError(e);
+                throw new RuntimeException(e);
             }
         }
     }

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

@@ -304,8 +304,7 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
                     // todo remove the    toByteArray(..)
                     writeValueUsingDelegate(map -> {
                         final byte[] key = toByteArray(valueIn);
-                        final byte[] value = map.get(key);
-                        return value;
+                        return map.get(key);
                     });
                     return;
                 }
@@ -316,9 +315,11 @@ public class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers>
 
                         final Params[] params = getAndPut.params();
 
-                        MapWireHandler.this.writeValue(b -> b.put(
-                                wire.read(params[0]).bytes(),
-                                wire.read(params[1]).bytes()));
+                        MapWireHandler.this.writeValue(b -> {
+                            byte[] k = wire.read(params[0]).bytes();
+                            byte[] v = wire.read(params[1]).bytes();
+                            return b.put(k, v);
+                        });
 
                     });
 

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

@@ -23,7 +23,6 @@ import net.openhft.chronicle.hash.hashing.LongHashFunction;
 import net.openhft.chronicle.hash.impl.ContextFactory;
 import net.openhft.chronicle.hash.impl.HashContext;
 import net.openhft.chronicle.hash.replication.AbstractReplication;
-import net.openhft.chronicle.hash.replication.LateUpdateException;
 import net.openhft.chronicle.hash.replication.TimeProvider;
 import net.openhft.chronicle.hash.serialization.BytesReader;
 import net.openhft.chronicle.hash.serialization.internal.BytesBytesInterop;
@@ -1009,7 +1008,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
                 (ReplicatedContext<K, KI, MKI, V, VI, MVI>) mapContext();
 
         // records the current position of the cursor in the bitset
-        private long position = -1L;
+        private volatile long position = -1L;
 
         /**
          * @param bytes                the back the bitset, used to mark which entries have changed

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

@@ -724,8 +724,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
             entryWriter.entriesToBuffer(attached.remoteModificationIterator, key);
 
         try {
-            final int len = entryWriter.writeBufferToSocket(socketChannel,
-                    approxTime);
+            final int len = entryWriter.writeBufferToSocket(socketChannel, approxTime);
 
             if (len == -1)
                 socketChannel.close();
@@ -733,8 +732,11 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
             if (len > 0)
                 contemplateThrottleWrites(len);
 
-            if (!entryWriter.hasBytesToWrite() && !entryWriter.isWorkIncomplete())
-                // TURN OP_WRITE_OFF
+            if (!entryWriter.hasBytesToWrite()
+                    && !entryWriter.isWorkIncomplete()
+                    && hasNext(attached))
+                // if we have no more data to write to the socket then we will
+                // un-register OP_WRITE on the selector, until more data becomes available
                 key.interestOps(key.interestOps() & ~OP_WRITE);
         } catch (IOException e) {
             quietClose(key, e);
@@ -744,6 +746,11 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
         }
     }
 
+    private boolean hasNext(Attached attached) {
+        return attached.remoteModificationIterator != null
+                && !attached.remoteModificationIterator.hasNext();
+    }
+
     /**
      * called when the selector receives a OP_READ message
      */
@@ -1150,15 +1157,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
                     if (entrySize > largestEntrySoFar)
                         largestEntrySoFar = entrySize;
 
-
-                    if (!modificationIterator.hasNext()) {
-                        // if we have no more data to write to the socket then we will
-                        // un-register OP_WRITE on the selector, until more data becomes available
-                        if (in().position() == 0 && handShakingComplete)
-                            disableWrite(socketChannel, attached);
-
-                    }
-
                     // we've filled up the buffer lets give another channel a chance to send
                     // some data
                     if (in().remaining() <= largestEntrySoFar || in().position() >

src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java

@@ -1,11 +1,11 @@
 package net.openhft.chronicle.map;
 
+import org.junit.BeforeClass;
+import org.junit.Ignore;
 import org.junit.Test;
 
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
+import java.util.function.Consumer;
 
 import static org.junit.Assert.*;
 
@@ -13,17 +13,18 @@ import static org.junit.Assert.*;
  * JUnit test class to support {@link FilePerKeyMapTest}
  */
 public class FilePerKeyMapTest {
+    private static FilePerKeyMap map;
+    @BeforeClass
+    public static void createMap(){
+        map = new FilePerKeyMap("/tmp/filepermaptests");
+        //just in case it hasn't been cleared up last time
+        map.clear();
+    }
     /**
      * Testing all the methods of the map with simple tests.
      */
     @Test
     public void testMapMethods() {
-        //There are no entries in the map so null should be returned
-        FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests");
-
-        //just in case it hasn't been cleared up last time
-        map.clear();
-
         String val = map.put("one", "test1");
         assertEquals(val, null);
 
@@ -103,25 +104,11 @@ public class FilePerKeyMapTest {
             }
         }
 
-        try {
-            Thread.sleep(3000);
-        } catch (InterruptedException e) {
-             e.printStackTrace();
-        }
-
         map.clear();
-        map.close();
-
     }
 
     @Test
     public void perfTest(){
-        //There are no entries in the map so null should be returned
-        FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests");
-
-        //just in case it hasn't been cleared up last time
-        map.clear();
-
         StringBuilder sb = new StringBuilder();
         for (int i = 0; i < 2_000_000; i++) {
             sb.append('x');
@@ -139,32 +126,47 @@ public class FilePerKeyMapTest {
         for (int i = 0; i < iterations; i++) {
             map.put("big file", value);
         }
-        System.out.println("Time to update "+ iterations + " iterations " + (System.currentTimeMillis()-time));
+        System.out.println("Time to update " + iterations + " iterations " + (System.currentTimeMillis() - time));
 
         map.clear();
-        map.close();
     }
 
+    @Ignore //This doesn't work on all OS
     @Test
     public void eventTest() {
-        //There are no entries in the map so null should be returned
-        FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests");
+        final List<FPMEvent> events = new ArrayList<>();
 
-        //just in case it hasn't been cleared up last time
-        map.clear();
-
-        map.registerForEvents(e -> System.out.println(e));
+        Consumer<FPMEvent> fpmEventConsumer = (FPMEvent e) -> {
+            System.out.println(e);
+            events.add(e);
+        };
+        map.registerForEvents(fpmEventConsumer);
 
         map.put("one", "one");
+        map.put("one", "two");
+        map.remove("one");
         map.put("one", "one");
 
         try {
+            //Allow all events to be played through
             Thread.sleep(3000);
+            //Check the events
+            assertEquals(4, events.size());
+
+            assertEquals(FPMEvent.EventType.NEW,events.get(0).getEventType());
+            assertEquals(FPMEvent.EventType.UPDATE, events.get(1).getEventType());
+            assertEquals(FPMEvent.EventType.DELETE, events.get(2).getEventType());
+            assertEquals(FPMEvent.EventType.NEW, events.get(3).getEventType());
+
+            assertEquals(true, events.get(0).isProgrammatic());
+            assertEquals(true, events.get(1).isProgrammatic());
+            assertEquals(true, events.get(2).isProgrammatic());
+            assertEquals(true, events.get(3).isProgrammatic());
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
 
+        map.unregisterForEvents(fpmEventConsumer);
         map.clear();
-        map.close();
     }
 }

src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java

@@ -3,13 +3,18 @@ package net.openhft.chronicle.map;
 import net.openhft.chronicle.hash.replication.ReplicationChannel;
 import net.openhft.chronicle.hash.replication.ReplicationHub;
 import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
+import net.openhft.lang.model.constraints.MaxSize;
+import net.openhft.lang.values.DoubleArray;
+import org.junit.Assert;
 import org.junit.Test;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.concurrent.TimeUnit;
 
+import static net.openhft.chronicle.map.ChronicleMapBuilder.of;
 import static org.junit.Assert.assertEquals;
 
 /**
@@ -121,4 +126,224 @@ public class TestEntriesSpanningSeveralChunksReplicationHub {
         favoriteColourServer1.close();
     }
 
+
+    public static final int NUMBER_OF_CHANNELS = 1000;
+
+    @Test(timeout = 100000)
+    public void test1000Channels()
+            throws IOException, InterruptedException {
+
+
+        final HashMap<Short, ChronicleMap> map1 = new HashMap<>();
+        final HashMap<Short, ChronicleMap> map2 = new HashMap<>();
+
+
+        // server 1 with  identifier = 1
+        {
+            ChronicleMapBuilder<CharSequence, Short> builder =
+                    of(CharSequence.class, Short.class)
+                            .averageKeySize(10)
+                            .averageValueSize(100)
+                            .entries(1000);
+
+            byte identifier = (byte) 1;
+
+            TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
+                    .of(8053);
+
+            ReplicationHub hubOnServer1 = ReplicationHub.builder()
+                    .tcpTransportAndNetwork(tcpConfig)
+                    .maxNumberOfChannels(NUMBER_OF_CHANNELS)
+                    .createWithId(identifier);
+
+            for (short channelId = 1; channelId < NUMBER_OF_CHANNELS; channelId++) {
+                ReplicationChannel channel = hubOnServer1.createChannel(channelId);
+                ChronicleMap map = builder.instance().replicatedViaChannel(channel).create();
+
+                map.put("key", channelId);
+                map1.put(channelId, map);
+
+            }
+        }
+
+
+        {
+            ChronicleMapBuilder<CharSequence, Short> builder =
+                    of(CharSequence.class, Short.class)
+                            .averageKeySize(10)
+                            .averageValueSize(100)
+                            .entries(1000);
+
+            byte identifier = (byte) 2;
+
+            TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
+                    .of(8054, new InetSocketAddress("localhost", 8053));
+
+            ReplicationHub hubOnServer1 = ReplicationHub.builder()
+                    .tcpTransportAndNetwork(tcpConfig)
+                    .maxNumberOfChannels(NUMBER_OF_CHANNELS)
+                    .createWithId(identifier);
+
+            for (short channelId = 1; channelId < NUMBER_OF_CHANNELS; channelId++) {
+                ReplicationChannel channel = hubOnServer1.createChannel(channelId);
+                ChronicleMap map = builder.instance().replicatedViaChannel(channel).create();
+
+                map.put("key", channelId);
+                map2.put(channelId, map);
+
+            }
+        }
+
+        // check that the maps are the same, if the are not the same the test will timeout
+        // we have to give a few seconds for the replication to propagate
+        for (short channelId = 1; channelId < NUMBER_OF_CHANNELS; channelId++) {
+            // allow time for the recompilation to resolve
+
+            final ChronicleMap v1 = map1.get(channelId);
+            final ChronicleMap v2 = map2.get(channelId);
+
+            for (; ; ) {
+
+
+                if (v1 == null || v2 == null) {
+                    Thread.sleep(10);
+                    continue;
+                }
+
+                if (v1.isEmpty() || v2.isEmpty()) {
+                    Thread.sleep(10);
+                    continue;
+                }
+
+                if (!v1.equals(v2)) {
+                    Thread.sleep(10);
+                    continue;
+                }
+
+
+                Assert.assertEquals(channelId, v1.get("key"));
+                Assert.assertEquals(channelId, v2.get("key"));
+
+                break;
+            }
+
+        }
+
+        // close
+
+        for (short channelId = 1; channelId < NUMBER_OF_CHANNELS; channelId++) {
+
+            final ChronicleMap v1 = map1.get(channelId);
+            if (v1 != null)
+                v1.close();
+
+            final ChronicleMap v2 = map2.get(channelId);
+            if (v2 != null)
+                v2.close();
+        }
+    }
+
+    @Test
+    public void testChannelWithDifferentTypes() throws IOException, InterruptedException {
+
+
+        // server 1  - in this example only server 1 write any data
+        ChronicleMap<CharSequence, CharSequence> charMap1;
+        ChronicleMap<Integer, Integer> intMap1;
+        {
+            ChronicleMapBuilder<Integer, Integer> intMapBuilder =
+                    of(Integer.class, Integer.class)
+                            .entries(1000);
+
+            ChronicleMapBuilder<CharSequence, CharSequence> charMapBuilder =
+                    of(CharSequence.class, CharSequence.class)
+                            .averageKeySize(10)
+                            .averageValueSize(100)
+                            .entries(1000);
+
+            byte identifier = (byte) 1;
+
+            TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8025);
+
+            ReplicationHub hubOnServer1 = ReplicationHub.builder()
+                    .tcpTransportAndNetwork(tcpConfig)
+                    .maxNumberOfChannels(NUMBER_OF_CHANNELS)
+                    .createWithId(identifier);
+
+
+            ReplicationChannel channel1 = hubOnServer1.createChannel(1);
+
+            intMap1 = intMapBuilder.instance()
+                    .replicatedViaChannel(channel1)
+                    .create();
+
+            ReplicationChannel channel2 = hubOnServer1.createChannel(2);
+
+            charMap1 = charMapBuilder.instance()
+                    .replicatedViaChannel
+                            (channel2)
+                    .create();
+
+            intMap1.put(1, 1);
+            charMap1.put("Hello", "World");
+
+        }
+
+
+        // server 2
+        ChronicleMap<CharSequence, CharSequence> charMap2;
+        ChronicleMap<Integer, Integer> intMap2;
+
+        {
+            ChronicleMapBuilder<Integer, Integer> intMapBuilder =
+                    of(Integer.class, Integer.class)
+                            .entries(1000);
+
+            ChronicleMapBuilder<CharSequence, CharSequence> charMapBuilder =
+                    of(CharSequence.class, CharSequence.class)
+                            .averageKeySize(10)
+                            .averageValueSize(100)
+                            .entries(1000);
+
+            byte identifier = (byte) 2;
+
+            TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8026, new
+                    InetSocketAddress("localhost", 8025));  // change the localhost to your server
+
+            ReplicationHub hubOnServer1 = ReplicationHub.builder()
+                    .tcpTransportAndNetwork(tcpConfig)
+                    .maxNumberOfChannels(NUMBER_OF_CHANNELS)
+                    .createWithId(identifier);
+
+
+            ReplicationChannel channel1 = hubOnServer1.createChannel(1);
+
+            intMap2 = intMapBuilder.instance()
+                    .replicatedViaChannel(channel1)
+                    .create();
+
+            ReplicationChannel channel2 = hubOnServer1.createChannel(2);
+
+            charMap2 = charMapBuilder.instance().replicatedViaChannel
+                    (channel2)
+                    .create();
+
+        }
+
+        // allow time for the recompilation to resolve
+        for (int t = 0; t < 3500; t++) {
+            if (intMap1.equals(intMap2) &&
+                    charMap1.equals(charMap2))
+                break;
+            Thread.sleep(1);
+        }
+
+        Assert.assertEquals(intMap1,intMap2);
+        Assert.assertEquals(charMap1,charMap2);
+
+    }
+
+
+
 }
+

src/test/java/net/openhft/chronicle/map/TestReplication.java

@@ -0,0 +1,57 @@
+package net.openhft.chronicle.map;
+
+import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
+import net.openhft.chronicle.map.ChronicleMap;
+import net.openhft.chronicle.map.ChronicleMapBuilder;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Created by Rob Austin
+ */
+public class TestReplication {
+
+
+    @Test
+    public void testAllDataGetsReplicated() throws InterruptedException {
+
+
+        TcpTransportAndNetworkConfig tcpConfigServer1 =
+                TcpTransportAndNetworkConfig.of(8082);
+
+        TcpTransportAndNetworkConfig tcpConfigServer2 =
+                TcpTransportAndNetworkConfig.of(8083, new InetSocketAddress("localhost",
+                        8082));
+
+        final ChronicleMap<Integer, Integer> map2 = ChronicleMapBuilder.of(Integer.class,
+                Integer.class)
+                .replication((byte) 2, tcpConfigServer2)
+                .create();
+
+
+        final ChronicleMap<Integer, Integer> map1 = ChronicleMapBuilder.of(Integer.class,
+                Integer.class)
+                .replication((byte) 3, tcpConfigServer1)
+                .create();
+
+        for (int i = 0; i < 70000; i++) {
+            map1.put(i, i);
+        }
+
+        for (int i = 0; i < 10; i++) {
+
+            Thread.sleep(100);
+            System.out.println(map2.size());
+        }
+
+
+        Assert.assertEquals(map1.size(),map2.size());
+
+    }
+
+
+    
+
+}