From 4743f9b870276f7b2452b33162a6f79f5a894242 Mon Sep 17 00:00:00 2001 From: danielshaya Date: Mon, 27 Apr 2015 10:30:15 +0100 Subject: [PATCH 01/11] Initial commit for FilePerMap implementation. --- .../openhft/chronicle/map/FilePerKeyMap.java | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java index 592c67ab4..b38f6e854 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java @@ -13,17 +13,26 @@ 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 key is the file name and the value + * is the contents of the file. This map will only handle String's. + *

+ * 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. */ public class FilePerKeyMap implements Map { - private final String dir; private final Path dirPath; private final Map lastModifiedByProgram = new ConcurrentHashMap<>(); private final List> 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 +68,6 @@ public class FilePerKeyMap implements Map { @Override public boolean containsKey(Object key) { - return getFiles().anyMatch(p->p.getFileName().toString().equals(key)); } @@ -87,10 +95,9 @@ public class FilePerKeyMap implements Map { 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 +139,7 @@ public class FilePerKeyMap implements Map { try { return Files.walk(dirPath).filter(p -> !Files.isDirectory(p)); } catch (IOException e) { - throw new AssertionError(e); + throw new RuntimeException(e); } } @@ -142,7 +149,7 @@ public class FilePerKeyMap implements Map { try { existingValue = Files.readAllLines(path).stream().collect(Collectors.joining()); } catch (IOException e) { - throw new AssertionError(e); + throw new RuntimeException(e); } } return existingValue; @@ -208,7 +215,6 @@ public class FilePerKeyMap implements Map { if (key != null ? !key.equals(fpmEntry.key) : fpmEntry.key != null) return false; return !(value != null ? !value.equals(fpmEntry.value) : fpmEntry.value != null); - } @Override From 17cc8a1f2d2c3b8bde2a28efe230a384e36677ed Mon Sep 17 00:00:00 2001 From: daniel Date: Mon, 27 Apr 2015 12:28:03 +0100 Subject: [PATCH 02/11] Changes to FilePerKeyMap made after testing on Linux. --- .../openhft/chronicle/map/FilePerKeyMap.java | 24 +++++-- .../chronicle/map/FilePerKeyMapTest.java | 68 +++++++++---------- 2 files changed, 53 insertions(+), 39 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java index b38f6e854..fcc60f61e 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java @@ -29,6 +29,7 @@ import java.util.stream.Stream; public class FilePerKeyMap implements Map { private final Path dirPath; private final Map lastModifiedByProgram = new ConcurrentHashMap<>(); + private final Map lastUpdate = new ConcurrentHashMap<>(); private final List> listeners = new ArrayList<>(); private final FPMWatcher fileFpmWatcher = new FPMWatcher(); @@ -147,7 +148,7 @@ public class FilePerKeyMap implements Map { 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 RuntimeException(e); } @@ -242,21 +243,24 @@ public class FilePerKeyMap implements Map { try { key = watcher.take(); for (WatchEvent event : key.pollEvents()) { - // get event type WatchEvent.Kind kind = event.kind(); // get file name WatchEvent ev = (WatchEvent) 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 { @@ -264,15 +268,25 @@ public class FilePerKeyMap implements Map { } } 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 { @@ -287,7 +301,7 @@ public class FilePerKeyMap implements Map { } } } catch (IOException e) { - throw new AssertionError(e); + throw new RuntimeException(e); } } } diff --git a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java index 56c75ae50..47434f081 100644 --- a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java @@ -1,11 +1,10 @@ package net.openhft.chronicle.map; +import org.junit.BeforeClass; 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 +12,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 +103,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 +125,46 @@ 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(); } @Test public void eventTest() { - //There are no entries in the map so null should be returned - FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests"); + final List events = new ArrayList<>(); - //just in case it hasn't been cleared up last time - map.clear(); - - map.registerForEvents(e -> System.out.println(e)); + Consumer 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 { - Thread.sleep(3000); + //Allow all events to be played through + Thread.sleep(1000); + //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(); } } From 0dadb1bdcb72bf621c01d535a1fb0bb202623139 Mon Sep 17 00:00:00 2001 From: danielshaya Date: Mon, 27 Apr 2015 12:34:51 +0100 Subject: [PATCH 03/11] Ignore test that doesn't work on all OS platforms. --- .../java/net/openhft/chronicle/map/FilePerKeyMapTest.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java index 47434f081..21279d6ba 100644 --- a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java @@ -1,6 +1,7 @@ package net.openhft.chronicle.map; import org.junit.BeforeClass; +import org.junit.Ignore; import org.junit.Test; import java.util.*; @@ -130,6 +131,7 @@ public class FilePerKeyMapTest { map.clear(); } + @Ignore //This doesn't work on all OS @Test public void eventTest() { final List events = new ArrayList<>(); @@ -147,7 +149,7 @@ public class FilePerKeyMapTest { try { //Allow all events to be played through - Thread.sleep(1000); + Thread.sleep(3000); //Check the events assertEquals(4, events.size()); From 24896ee3c4e9a8c953b30da775b94565e0c249f5 Mon Sep 17 00:00:00 2001 From: danielshaya Date: Mon, 27 Apr 2015 14:04:02 +0100 Subject: [PATCH 04/11] Added to JavaDoc. --- .../net/openhft/chronicle/map/FilePerKeyMap.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java index fcc60f61e..299995227 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java @@ -14,8 +14,8 @@ import java.util.stream.Stream; /** * A {@link Map} implementation that stores each entry as a file in a - * directory. The key is the file name and the value - * is the contents of the file. This map will only handle String's. + * directory. The key is the file name and the value + * is the contents of the file. This map will only handle String's. *

* 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 @@ -25,6 +25,14 @@ import java.util.stream.Stream; * (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. + *

+ * 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. + *

+ * 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 { private final Path dirPath; From b7e500382d72597dbeec0cffcbde5b70a4b8c44f Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 27 Apr 2015 14:50:28 +0100 Subject: [PATCH 05/11] HCOLL-325 Issue when bootstrapping with a large number of entries, and removed a few @Ignores on tests that should work. --- .../chronicle/map/ReplicatedChronicleMap.java | 3 +-- .../openhft/chronicle/map/TcpReplicator.java | 24 +++++++++---------- 2 files changed, 12 insertions(+), 15 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 0aeed758c..60df821a5 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/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) 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 diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index b8e81bd93..4c4eb5ff6 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -724,8 +724,7 @@ final class TcpReplicator 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 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 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 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() > From ff9f0e32af21ee6efb498f91aa5d3ed3d99213a9 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 27 Apr 2015 19:15:51 +0100 Subject: [PATCH 06/11] CE-24 Create marshal for Map type ( in wire ) --- .../java/net/openhft/chronicle/map/EngineMap.java | 5 ++++- .../net/openhft/chronicle/map/MapWireHandler.java | 11 ++++++----- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/EngineMap.java b/src/main/java/net/openhft/chronicle/map/EngineMap.java index 0a25773df..facd20a1b 100644 --- a/src/main/java/net/openhft/chronicle/map/EngineMap.java +++ b/src/main/java/net/openhft/chronicle/map/EngineMap.java @@ -166,7 +166,10 @@ public class EngineMap implements ChronicleMap { @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 diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 81cba95b4..45d5d3177 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -304,8 +304,7 @@ public class MapWireHandler implements WireHandler, Consumer // 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 implements WireHandler, Consumer 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); + }); }); From dadc3db8beb0fd1c7936fa290ae50c5f1d950740 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 27 Apr 2015 19:24:37 +0100 Subject: [PATCH 07/11] HCOLL-325 added test --- .../chronicle/map/TestReplication.java | 57 +++++++++++++++++++ 1 file changed, 57 insertions(+) create mode 100644 src/test/java/net/openhft/chronicle/map/TestReplication.java diff --git a/src/test/java/net/openhft/chronicle/map/TestReplication.java b/src/test/java/net/openhft/chronicle/map/TestReplication.java new file mode 100644 index 000000000..e967af61c --- /dev/null +++ b/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 map2 = ChronicleMapBuilder.of(Integer.class, + Integer.class) + .replication((byte) 2, tcpConfigServer2) + .create(); + + + final ChronicleMap 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()); + + } + + + + +} From e9f475a3ed61c4e4c5ff9ebf9d13f6dfa4c671f4 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 28 Apr 2015 11:45:01 +0100 Subject: [PATCH 08/11] added a test to check that we can support a thousand channels --- ...esSpanningSeveralChunksReplicationHub.java | 121 ++++++++++++++++++ 1 file changed, 121 insertions(+) diff --git a/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java b/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java index 7bc23c9d8..90a65da22 100644 --- a/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java +++ b/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java @@ -3,13 +3,16 @@ 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 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 +124,122 @@ public class TestEntriesSpanningSeveralChunksReplicationHub { favoriteColourServer1.close(); } + + public static final int NUMBER_OF_CHANNELS = 1000; + + @Test(timeout = 50000) + public void test1000Channels() + throws IOException, InterruptedException { + + + final HashMap map1 = new HashMap<>(); + final HashMap map2 = new HashMap<>(); + + + // server 1 with identifier = 1 + { + ChronicleMapBuilder builder = + of(CharSequence.class, Short.class) + .averageKeySize(10) + .averageValueSize(100) + .entries(1000); + + byte identifier = (byte) 1; + + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .of(8021); + + 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 builder = + of(CharSequence.class, Short.class) + .averageKeySize(10) + .averageValueSize(100) + .entries(1000); + + byte identifier = (byte) 2; + + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .of(8022, new InetSocketAddress("localhost", 8021)); + + 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(); + } + } + } + From f64f734b8797c22c599aec529a0e4d862d3dbba7 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 28 Apr 2015 12:04:50 +0100 Subject: [PATCH 09/11] added an example of channels with different types in a single replication hub --- ...esSpanningSeveralChunksReplicationHub.java | 102 +++++++++++++++++- 1 file changed, 101 insertions(+), 1 deletion(-) diff --git a/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java b/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java index 90a65da22..c54c37b75 100644 --- a/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java +++ b/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java @@ -127,7 +127,7 @@ public class TestEntriesSpanningSeveralChunksReplicationHub { public static final int NUMBER_OF_CHANNELS = 1000; - @Test(timeout = 50000) + @Test(timeout = 100000) public void test1000Channels() throws IOException, InterruptedException { @@ -241,5 +241,105 @@ public class TestEntriesSpanningSeveralChunksReplicationHub { } } + @Test + public void testChannelWithDifferentTypes() throws IOException, InterruptedException { + + + // server 1 - in this example only server 1 write any data + ChronicleMap charMap1; + ChronicleMap intMap1; + { + ChronicleMapBuilder intMapBuilder = + of(Integer.class, Integer.class) + .entries(1000); + + ChronicleMapBuilder charMapBuilder = + of(CharSequence.class, CharSequence.class) + .averageKeySize(10) + .averageValueSize(100) + .entries(1000); + + byte identifier = (byte) 1; + + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8021); + + 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 charMap2; + ChronicleMap intMap2; + + { + ChronicleMapBuilder intMapBuilder = + of(Integer.class, Integer.class) + .entries(1000); + + ChronicleMapBuilder charMapBuilder = + of(CharSequence.class, CharSequence.class) + .averageKeySize(10) + .averageValueSize(100) + .entries(1000); + + byte identifier = (byte) 2; + + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8022, new + InetSocketAddress("localhost", 8021)); // 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); + + } + } From 4c22aaf64ae047a663f40e7933e7737798791699 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 28 Apr 2015 13:03:03 +0100 Subject: [PATCH 10/11] changed ports --- ...TestEntriesSpanningSeveralChunksReplicationHub.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java b/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java index c54c37b75..b79a4e713 100644 --- a/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java +++ b/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java @@ -147,7 +147,7 @@ public class TestEntriesSpanningSeveralChunksReplicationHub { byte identifier = (byte) 1; TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig - .of(8021); + .of(8023); ReplicationHub hubOnServer1 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) @@ -175,7 +175,7 @@ public class TestEntriesSpanningSeveralChunksReplicationHub { byte identifier = (byte) 2; TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig - .of(8022, new InetSocketAddress("localhost", 8021)); + .of(8024, new InetSocketAddress("localhost", 8023)); ReplicationHub hubOnServer1 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) @@ -261,7 +261,7 @@ public class TestEntriesSpanningSeveralChunksReplicationHub { byte identifier = (byte) 1; - TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8021); + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8025); ReplicationHub hubOnServer1 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) @@ -305,8 +305,8 @@ public class TestEntriesSpanningSeveralChunksReplicationHub { byte identifier = (byte) 2; - TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8022, new - InetSocketAddress("localhost", 8021)); // change the localhost to your server + TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(8026, new + InetSocketAddress("localhost", 8025)); // change the localhost to your server ReplicationHub hubOnServer1 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) From e4e8efe295f1001c6091fc50f599225d2977649d Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 28 Apr 2015 13:16:40 +0100 Subject: [PATCH 11/11] changed ports --- .../TestEntriesSpanningSeveralChunksReplicationHub.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java b/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java index b79a4e713..cf55fa3f7 100644 --- a/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java +++ b/src/test/java/net/openhft/chronicle/map/TestEntriesSpanningSeveralChunksReplicationHub.java @@ -3,6 +3,8 @@ 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; @@ -147,7 +149,7 @@ public class TestEntriesSpanningSeveralChunksReplicationHub { byte identifier = (byte) 1; TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig - .of(8023); + .of(8053); ReplicationHub hubOnServer1 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) @@ -175,7 +177,7 @@ public class TestEntriesSpanningSeveralChunksReplicationHub { byte identifier = (byte) 2; TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig - .of(8024, new InetSocketAddress("localhost", 8023)); + .of(8054, new InetSocketAddress("localhost", 8053)); ReplicationHub hubOnServer1 = ReplicationHub.builder() .tcpTransportAndNetwork(tcpConfig) @@ -341,5 +343,7 @@ public class TestEntriesSpanningSeveralChunksReplicationHub { } + + }