From f79029f280599150f6f0178b8e472c4ea03f645d Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 11:04:07 +0000 Subject: [PATCH 01/24] removed getLater(0 putLater() and removeLater() because the stateless client is now mulithreaded --- .../openhft/chronicle/map/ChronicleMap.java | 62 ------------ .../chronicle/map/StatelessChronicleMap.java | 34 ------- .../chronicle/map/VanillaChronicleMap.java | 99 ------------------- .../chronicle/map/CHMUseCasesTest.java | 48 ++------- .../chronicle/map/ReplicationCheckingMap.java | 16 --- .../chronicle/map/StatelessClientTest.java | 53 ---------- .../jrs166/map/StatelessChronicleMapTest.java | 16 --- 7 files changed, 6 insertions(+), 322 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java index e0cef2c72..c7e3764fd 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java @@ -211,68 +211,6 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { */ R updateForKey(K key, @NotNull Mutator mutator); - /** - * Returns a {@link java.util.concurrent.Future} containing the value to which the specified key - * is mapped, or {@code null} if this map contains no mapping for the key. This method behaves - * the same as {@link java.util.Map#get(java.lang.Object)} yet wraps result in a future a {@link - * java.util.concurrent.Future}. - * - * Calling this method may or may not block until the operating is complete depending on the - * implementation. - * - * @param key the key whose associated value is to be returned in the future - * @return a future containing the value to which the specified key is mapped after this method - * call, or {@code null} if no value is mapped - * @see java.util.Map#get(java.lang.Object) - * @see java.util.concurrent.Future - */ - Future getLater(@NotNull K key); - - /** - * Associates the specified value with the specified key in this map (optional operation). If - * the map previously contained a mapping for the key, the old value is replaced by the - * specified value. (A map m is said to contain a mapping for a key k if and - * only if {@link #containsKey(Object) m.containsKey(k)} would return true.) - * - * Calling this method may or may not block until the operating is complete depending on the - * implementation. - * - * @param key key with which the specified value is to be associated - * @param value value to be associated with the specified key * @return a future containing the - * previous value associated with key, or null if there was no - * mapping for key. (A null return can also indicate that the map - * previously associated null with key, if the implementation - * supports null values.) - * @return a future containing either, the previous value associated with key, or - * null if there was no mapping for key. - * @see java.util.Map#put(java.lang.Object, java.lang.Object) - * @see java.util.concurrent.Future - */ - Future putLater(@NotNull K key, @NotNull V value); - - /** - * Removes the mapping for a key from this map if it is present (optional operation). More - * formally, if this map contains a mapping from key k to value v such that - * (key==null ? k==null : key.equals(k)), that mapping is removed. (The map can - * contain at most one such mapping.) - * - *

Returns a future containing, either the value to which this map previously associated the - * key, or null if the map contained no mapping for the key. - * - * This method behaves the same as {@link java.util.Map#remove(java.lang.Object)} yet wraps - * result in a future a {@link java.util.concurrent.Future}. - * - * Calling this method may or may not block until the operating is complete depending on the - * implementation. - * - * @param key key whose mapping is to be removed from the map - * @return a future containing either, the previous value associated with key, or - * null if there was no mapping for key. - * @see java.util.Map#remove(java.lang.Object) - * @see java.util.concurrent.Future - */ - Future removeLater(@NotNull K key); - /** * exports all the entries to a {@link java.io.File} storing them in JSON format, an attempt is * made where possible to use standard java serialisation and keep the data human readable, data diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 59e2dbf3a..29fc940f8 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -157,40 +157,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - @NotNull - @Override - public Future getLater(@NotNull final K key) { - return lazyExecutorService().submit(new Callable() { - @Nullable - @Override - public V call() throws Exception { - return StatelessChronicleMap.this.get(key); - } - }); - } - - @NotNull - @Override - public Future putLater(@NotNull final K key, @NotNull final V value) { - return lazyExecutorService().submit(new Callable() { - @Nullable - @Override - public V call() throws Exception { - return StatelessChronicleMap.this.put(key, value); - } - }); - } - - @Override - public Future removeLater(@NotNull final K key) { - return lazyExecutorService().submit(new Callable() { - @Override - public V call() throws Exception { - return StatelessChronicleMap.this.remove(key); - } - }); - } - @Override public void getAll(File toFile) { throw new UnsupportedOperationException(); diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 2670bec71..d83b83753 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -50,10 +50,7 @@ import java.lang.reflect.Array; import java.nio.ByteOrder; import java.nio.channels.FileChannel; import java.util.*; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import static java.lang.Class.forName; import static java.lang.Long.numberOfTrailingZeros; @@ -569,102 +566,6 @@ class VanillaChronicleMap, } } - @Override - public Future getLater(@NotNull K key) { - final V v = get(key); - return new Future() { - @Override - public boolean cancel(boolean mayInterruptIfRunning) { - return false; - } - - @Override - public boolean isCancelled() { - return false; - } - - @Override - public boolean isDone() { - return true; - } - - @Override - public V get() throws InterruptedException, ExecutionException { - return v; - } - - @Override - public V get(long timeout, @NotNull TimeUnit unit) - throws InterruptedException, ExecutionException, TimeoutException { - return v; - } - }; - } - - @Override - public Future putLater(@NotNull K key, @NotNull V value) { - final V v = put(key, value); - return new Future() { - @Override - public boolean cancel(boolean mayInterruptIfRunning) { - return false; - } - - @Override - public boolean isCancelled() { - return false; - } - - @Override - public boolean isDone() { - return true; - } - - @Override - public V get() throws InterruptedException, ExecutionException { - return v; - } - - @Override - public V get(long timeout, TimeUnit unit) - throws InterruptedException, ExecutionException, TimeoutException { - return v; - } - }; - } - - @Override - public Future removeLater(@NotNull K key) { - final V v = remove(key); - return new Future() { - @Override - public boolean cancel(boolean mayInterruptIfRunning) { - return false; - } - - @Override - public boolean isCancelled() { - return false; - } - - @Override - public boolean isDone() { - return true; - } - - @Override - public V get() throws InterruptedException, ExecutionException { - return v; - } - - @Override - public V get(long timeout, TimeUnit unit) - throws InterruptedException, ExecutionException, TimeoutException { - return v; - } - }; - } - @Override public synchronized void getAll(File toFile) throws IOException { diff --git a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java index 6c98eb444..98eab7796 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java +++ b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java @@ -36,7 +36,7 @@ public class CHMUseCasesTest { */ try (ChronicleMap map = ChronicleMapBuilder .of(String.class, String.class) - .checkSerializedValues() // for testng purposes only + .checkSerializedValues() // for testing purposes only .create()) { map.put("Hello", "World"); assertEquals("World", map.get("Hello")); @@ -65,10 +65,6 @@ public class CHMUseCasesTest { // TODO replace with the specific exception. } - assertEquals(null, map.putLater("Bye", "For now").get()); - assertEquals("For now", map.getLater("Bye").get()); - assertEquals("For now", map.removeLater("Bye").get()); - assertEquals(null, map.removeLater("Bye").get()); } } @@ -81,7 +77,7 @@ public class CHMUseCasesTest { public void testCharSequenceCharSequenceMap() throws ExecutionException, InterruptedException { try (ChronicleMap map = ChronicleMapBuilder .of(CharSequence.class, CharSequence.class) - .checkSerializedValues() // for testng purposes only + .checkSerializedValues() // for testing purposes only .create()) { map.put("Hello", "World"); StringBuilder key = new StringBuilder(); @@ -129,10 +125,6 @@ public class CHMUseCasesTest { } })); - assertEquals(null, map.putLater("Bye", "For now").get()); - assertEquals("For now", map.getLater("Bye").get().toString()); - assertEquals("For now", map.removeLater("Bye").get().toString()); - assertEquals(null, map.removeLater("Bye").get()); } } @@ -144,7 +136,7 @@ public class CHMUseCasesTest { public void testStringValueStringValueMap() { try (ChronicleMap map = ChronicleMapBuilder .of(StringValue.class, StringValue.class) - .checkSerializedValues() // for testng purposes only + .checkSerializedValues() // for testing purposes only .create()) { StringValue key1 = DataValueClasses.newDirectInstance(StringValue.class); StringValue key2 = DataValueClasses.newInstance(StringValue.class); @@ -283,10 +275,7 @@ public class CHMUseCasesTest { } catch (Exception todoMoreSpecificException) { } - assertEquals(null, map.putLater(3, 4).get()); - assertEquals((Integer) 4, map.getLater(3).get()); - assertEquals((Integer) 4, map.removeLater(3).get()); - assertEquals(null, map.removeLater(3).get()); + } } @@ -331,10 +320,7 @@ public class CHMUseCasesTest { } catch (Exception todoMoreSpecificException) { } - assertEquals(null, map.putLater(3L, 4L).get()); - assertEquals((Long) 4L, map.getLater(3L).get()); - assertEquals((Long) 4L, map.removeLater(3L).get()); - assertEquals(null, map.removeLater(3L).get()); + } } @@ -379,10 +365,7 @@ public class CHMUseCasesTest { } catch (Exception todoMoreSpecificException) { } - assertEquals(null, map.putLater(3.0, 4.0).get()); - assertEquals((Double) 4.0, map.getLater(3.0).get()); - assertEquals((Double) 4.0, map.removeLater(3.0).get()); - assertEquals(null, map.removeLater(3.0).get()); + } } @@ -430,10 +413,6 @@ public class CHMUseCasesTest { byte[] key3 = {3, 3, 3, 3}; byte[] value3 = {4, 4, 4, 4}; - assertEquals(null, map.putLater(key3, value3).get()); - assertTrue(Arrays.equals(value3, map.getLater(key3).get())); - assertTrue(Arrays.equals(value3, map.removeLater(key3).get())); - assertEquals(null, map.removeLater(key3).get()); } } @@ -521,13 +500,7 @@ public class CHMUseCasesTest { assertBBEquals(bb1, valueA); } - ByteBuffer key3 = ByteBuffer.wrap(new byte[]{3, 3, 3, 3}); - ByteBuffer value3 = ByteBuffer.wrap(new byte[]{4, 4, 4, 4}); - assertEquals(null, map.putLater(key3, value3).get()); - assertBBEquals(value3, map.getLater(key3).get()); - assertBBEquals(value3, map.removeLater(key3).get()); - assertEquals(null, map.removeLater(key3).get()); } } @@ -579,15 +552,6 @@ public class CHMUseCasesTest { assertBBEquals(ByteBuffer.wrap(new byte[]{12, 10, 11, 11}), map.get(key1)); - ByteBuffer key3 = ByteBuffer.wrap(new byte[]{3, 3, 3, 3}); - ByteBuffer value3 = ByteBuffer.allocateDirect(4); - value3.put(new byte[]{4, 4, 4, 4}); - value3.flip(); - - assertEquals(null, map.putLater(key3, value3).get()); - assertBBEquals(value3, map.getLater(key3).get()); - assertBBEquals(value3, map.removeLater(key3).get()); - assertEquals(null, map.removeLater(key3).get()); } } diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index 10292c16f..e70843d33 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -26,7 +26,6 @@ import java.util.Collection; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Future; /** * @author Rob Austin. @@ -228,21 +227,6 @@ public class ReplicationCheckingMap implements ChronicleMap { return map1.updateForKey(key, mutator); } - @Override - public Future getLater(@NotNull K key) { - throw new UnsupportedOperationException(); - } - - @Override - public Future putLater(@NotNull K key, @NotNull V value) { - throw new UnsupportedOperationException(); - } - - @Override - public Future removeLater(@NotNull K key) { - throw new UnsupportedOperationException(); - } - @Override public void getAll(File toFile) { throw new UnsupportedOperationException(); diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 66b664e45..1d4288b61 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -453,60 +453,7 @@ public class StatelessClientTest { } } - @Test(timeout = 10000) - public void testGetLater() throws IOException, - InterruptedException, ExecutionException { - try (ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) { - try (ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer - .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { - statelessMap.put(1, "some value"); - - assertEquals("some value", statelessMap.getLater(1).get()); - assertEquals(1, statelessMap.size()); - - statelessMap.remove(1); - - assertEquals(null, statelessMap.getLater(1).get()); - assertEquals(0, statelessMap.size()); - } - } - } - - @Test - public void testPutLater() throws IOException, - InterruptedException, ExecutionException { - - try (ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) { - try (ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer - .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { - CharSequence oldValue = statelessMap.putLater(1, "some value").get(); - assertEquals("some value", statelessMap.get(1)); - assertEquals(1, statelessMap.size()); - } - } - } - - @Test(timeout = 10000) - public void testRemoveLater() throws IOException, - InterruptedException, ExecutionException { - - try (ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) { - try (ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer - .class, CharSequence.class) - .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { - statelessMap.put(1, "some value"); - statelessMap.removeLater(1).get(); - assertEquals(null, statelessMap.get(1)); - assertEquals(0, statelessMap.size()); - } - } - } @Test(timeout = 10000) public void testGetAndEntryWeDontHave() throws IOException, diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java index 3aa530b2c..1c487acf1 100755 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java @@ -33,7 +33,6 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.*; -import java.util.concurrent.Future; import static org.junit.Assert.*; @@ -180,21 +179,6 @@ public class StatelessChronicleMapTest extends JSR166TestCase { return d.updateForKey(key, mutator); } - @Override - public Future getLater(@NotNull K key) { - return d.getLater(key); - } - - @Override - public Future putLater(@NotNull K key, @NotNull V value) { - return d.putLater(key, value); - } - - @Override - public Future removeLater(@NotNull K key) { - return d.removeLater(key); - } - @Override public void getAll(File toFile) throws IOException { d.getAll(toFile); From 5a8079f1e363847048b571bc4db89fb8dfcc115b Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 12:03:37 +0000 Subject: [PATCH 02/24] added "There is two key differences between Windows and Linux.." --- README.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/README.md b/README.md index cdbbc3dac..9f74ae368 100644 --- a/README.md +++ b/README.md @@ -295,6 +295,21 @@ To illustrate this with an example - On Ubuntu we can create a 100 TB chronicle `ls -l` say the process virtual size / file size is 100 TB, however the resident memory via `du` says the size is 71 MB after adding 10000 entries. You can see the size actually used with du. +There is two key differences between Windows and Linux + +- windows fails if you attempt to use more than 2^20 * 4 KB pages in a single mapping (4 GB). This doesn't fail when you map the region, rather as you use it up. We should check for this and prevent users mapping regions of more than 4 GB on some (?all?) windows. It may be this limitation doesn't apply to newer or server based versions, so this requires some testing. In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. It is possible Microsoft might fix this first. + +- windows allocates memory and disk eagerly. Linux allocates memory and disk lazily. MacOSX allocates memory lazily and disk eagerly. On windows, eager memory allocation means you can't map more than free memory, but it should reduce jitter when you use it. Chronicle Map allocates head room which is a waste on Windows (Linux's sparse allocation means the head room has little impact) We could add an option for linux to more eagerly allocate space/disk on startup. e.g. to a percentage like 50%. + +Linux systems see a performance degradation at around 200% of main memory size. + +There likely to be more subtle differences, but these are the main ones which have caused problems. + +For production systems, we recommend; +- on Windows smaller map sizes of less than 4 GB each, and less than 50% main memory in total. +- on Linux small to large maps of less than double main memory. e.g. if you have a 128 GB server, have less than 256 GB of maps on the server. +- on MacOSX, we have no specific recommendations. + ### Chronicle Map Interface The Chronicle Map interface adds a few methods above an beyond the standard ConcurrentMap, the ChronicleMapBuilder can also be used to return the ChronicleMap, see the example below : From bddefa7a70ee61a548add2f517d2ee855660c5d5 Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 12:05:57 +0000 Subject: [PATCH 03/24] added heading "How Operating Systems differ " --- README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/README.md b/README.md index 9f74ae368..9f89b37dd 100644 --- a/README.md +++ b/README.md @@ -295,6 +295,8 @@ To illustrate this with an example - On Ubuntu we can create a 100 TB chronicle `ls -l` say the process virtual size / file size is 100 TB, however the resident memory via `du` says the size is 71 MB after adding 10000 entries. You can see the size actually used with du. +### How Operating Systems differ + There is two key differences between Windows and Linux - windows fails if you attempt to use more than 2^20 * 4 KB pages in a single mapping (4 GB). This doesn't fail when you map the region, rather as you use it up. We should check for this and prevent users mapping regions of more than 4 GB on some (?all?) windows. It may be this limitation doesn't apply to newer or server based versions, so this requires some testing. In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. It is possible Microsoft might fix this first. From 1a43e5cc02032e454d86bbff3ceff27a1d3cdbfa Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 12:11:19 +0000 Subject: [PATCH 04/24] improved "How Operating Systems differ" --- README.md | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 9f89b37dd..62bb65cfc 100644 --- a/README.md +++ b/README.md @@ -299,14 +299,21 @@ says the size is 71 MB after adding 10000 entries. You can see the size actually There is two key differences between Windows and Linux -- windows fails if you attempt to use more than 2^20 * 4 KB pages in a single mapping (4 GB). This doesn't fail when you map the region, rather as you use it up. We should check for this and prevent users mapping regions of more than 4 GB on some (?all?) windows. It may be this limitation doesn't apply to newer or server based versions, so this requires some testing. In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. It is possible Microsoft might fix this first. +- Windows fails if you attempt to use more than 2^20 * 4 KB pages in a single mapping (4 GB). +This doesn't fail when you map the region, rather as you use it up. +It may be this limitation doesn't apply to newer or server based versions. +In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. -- windows allocates memory and disk eagerly. Linux allocates memory and disk lazily. MacOSX allocates memory lazily and disk eagerly. On windows, eager memory allocation means you can't map more than free memory, but it should reduce jitter when you use it. Chronicle Map allocates head room which is a waste on Windows (Linux's sparse allocation means the head room has little impact) We could add an option for linux to more eagerly allocate space/disk on startup. e.g. to a percentage like 50%. +- Windows allocates memory and disk eagerly. +- Linux allocates memory and disk lazily. +- MacOSX allocates memory lazily and disk eagerly. + +On Windows, eager memory allocation means you can't map more than free memory, but it should reduce jitter when you use it. +Chronicle Map allocates head room which is a waste on Windows (Linux's sparse allocation means the head room has +little impact). Linux systems see a performance degradation at around 200% of main memory size. -There likely to be more subtle differences, but these are the main ones which have caused problems. - For production systems, we recommend; - on Windows smaller map sizes of less than 4 GB each, and less than 50% main memory in total. - on Linux small to large maps of less than double main memory. e.g. if you have a 128 GB server, have less than 256 GB of maps on the server. From b8d7e874765a37e01d42b30ea7fbd59cd5416a1e Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 12:12:42 +0000 Subject: [PATCH 05/24] improved "How Operating Systems differ" --- README.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/README.md b/README.md index 62bb65cfc..6dac3fe0a 100644 --- a/README.md +++ b/README.md @@ -302,9 +302,7 @@ There is two key differences between Windows and Linux - Windows fails if you attempt to use more than 2^20 * 4 KB pages in a single mapping (4 GB). This doesn't fail when you map the region, rather as you use it up. It may be this limitation doesn't apply to newer or server based versions. -In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. - -- Windows allocates memory and disk eagerly. +In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. Windows allocates memory and disk eagerly. - Linux allocates memory and disk lazily. - MacOSX allocates memory lazily and disk eagerly. From a56f8e6aef1320547c3bbc0a99babed86f13acd8 Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 12:15:32 +0000 Subject: [PATCH 06/24] improved "How Operating Systems differ" --- README.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/README.md b/README.md index 6dac3fe0a..f8032281d 100644 --- a/README.md +++ b/README.md @@ -299,8 +299,7 @@ says the size is 71 MB after adding 10000 entries. You can see the size actually There is two key differences between Windows and Linux -- Windows fails if you attempt to use more than 2^20 * 4 KB pages in a single mapping (4 GB). -This doesn't fail when you map the region, rather as you use it up. +- Windows fails if you attempt to use more than 4 GB in a single mapping. This calculated as 4GB = 2^20 * 4 KB pages. This doesn't fail when you map the region, rather as you use it up. It may be this limitation doesn't apply to newer or server based versions. In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. Windows allocates memory and disk eagerly. - Linux allocates memory and disk lazily. From 35e619d0f3085bd6d02874714b7ddeea0751f280 Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 12:17:57 +0000 Subject: [PATCH 07/24] improved "How Operating Systems differ" --- README.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index f8032281d..c01a5bb47 100644 --- a/README.md +++ b/README.md @@ -299,9 +299,9 @@ says the size is 71 MB after adding 10000 entries. You can see the size actually There is two key differences between Windows and Linux -- Windows fails if you attempt to use more than 4 GB in a single mapping. This calculated as 4GB = 2^20 * 4 KB pages. This doesn't fail when you map the region, rather as you use it up. -It may be this limitation doesn't apply to newer or server based versions. -In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. Windows allocates memory and disk eagerly. +- Windows allocates memory and disk eagerly - Windows fails if you attempt to use more than 4 GB in a single mapping. This calculated as 4GB = 2^20 * 4 KB pages. This doesn't fail when you map the region, rather as you use it up. +It may be this limitation doesn't apply to newer or server based versions of Windows. +In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. - Linux allocates memory and disk lazily. - MacOSX allocates memory lazily and disk eagerly. From 291857e4d2da705e7ca49f265228486bfed4d5a2 Mon Sep 17 00:00:00 2001 From: leventov Date: Sun, 7 Dec 2014 20:19:28 +0700 Subject: [PATCH 08/24] Some debug output in TcpReplicator --- .../openhft/chronicle/map/TcpReplicator.java | 162 ++++++++++-------- 1 file changed, 88 insertions(+), 74 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 4bdad2b43..50bafefc5 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -974,24 +974,31 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // this can occur when new map are added to a channel final boolean handShakingComplete = attached.isHandShakingComplete(); - for (; ; ) { - final boolean wasDataRead = modificationIterator.nextEntry(entryCallback, 0); + int entriesWritten = 0; + try { + for (; ; entriesWritten++) { + final boolean wasDataRead = modificationIterator.nextEntry(entryCallback, 0); - if (!wasDataRead) { - // 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); + if (!wasDataRead) { + // 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); - return; + return; + } + + // we've filled up the buffer lets give another channel a chance to send + // some data + if (in.remaining() <= maxEntrySizeBytes) + return; + + // if we have space in the buffer to write more data and we just wrote data + // into the buffer then let try and write some more } - - // we've filled up the buffer lets give another channel a chance to send some data - if (in.remaining() <= maxEntrySizeBytes) - return; - - // if we have space in the buffer to write more data and we just wrote data into the - // buffer then let try and write some more + } finally { + if (LOG.isDebugEnabled()) + LOG.debug("Entries written: {}", entriesWritten); } } @@ -1152,82 +1159,89 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * @throws InterruptedException */ void entriesFromBuffer(@NotNull Attached attached, @NotNull SelectionKey key) { - for (; ; ) { - out.limit(in.position()); + int entriesRead = 0; + try { + for (; ; entriesRead++) { + out.limit(in.position()); - // its set to MIN_VALUE when it should be read again - if (state == NOT_SET) { - if (out.remaining() < SIZE_OF_SIZE + 1) - return; + // its set to MIN_VALUE when it should be read again + if (state == NOT_SET) { + if (out.remaining() < SIZE_OF_SIZE + 1) + return; - // state is used for both heartbeat and stateless - state = out.readByte(); - sizeInBytes = out.readInt(); + // state is used for both heartbeat and stateless + state = out.readByte(); + sizeInBytes = out.readInt(); - assert sizeInBytes >= 0; + assert sizeInBytes >= 0; - // if the buffer is too small to read this payload we will have to grow the - // size of the buffer - long requiredSize = sizeInBytes + SIZE_OF_SIZE + 1; - if (out.capacity() < requiredSize) { - attached.entryReader.resizeBuffer(requiredSize + HEADROOM); + // if the buffer is too small to read this payload we will have to grow the + // size of the buffer + long requiredSize = sizeInBytes + SIZE_OF_SIZE + 1; + if (out.capacity() < requiredSize) { + attached.entryReader.resizeBuffer(requiredSize + HEADROOM); + } + + // this is the : + // -- heartbeat if its 0 + // -- stateful update if its 1 + // -- the id of the stateful event + if (state == NOT_SET) + continue; } - // this is the : - // -- heartbeat if its 0 - // -- stateful update if its 1 - // -- the id of the stateful event - if (state == NOT_SET) - continue; - } + if (out.remaining() < sizeInBytes) { + return; + } - if (out.remaining() < sizeInBytes) { - return; - } + final long nextEntryPos = out.position() + sizeInBytes; + assert nextEntryPos > 0; + final long limit = out.limit(); + out.limit(nextEntryPos); - final long nextEntryPos = out.position() + sizeInBytes; - assert nextEntryPos > 0; - final long limit = out.limit(); - out.limit(nextEntryPos); + boolean isStatelessClient = (state != 1); - boolean isStatelessClient = (state != 1); + if (isStatelessClient) { + if (statelessServerConnector == null) { + LOG.error("", new IllegalArgumentException("received an event " + + "from a stateless map, stateless maps are not " + + "currently supported when using Chronicle Channels")); + } else { - if (isStatelessClient) { - if (statelessServerConnector == null) { - LOG.error("", new IllegalArgumentException("received an event " + - "from a stateless map, stateless maps are not " + - "currently supported when using Chronicle Channels")); - } else { + final Work futureWork = + statelessServerConnector.processStatelessEvent(state, + attached.entryWriter.in, attached.entryReader.out); - final Work futureWork = statelessServerConnector.processStatelessEvent(state, - attached.entryWriter.in, attached.entryReader.out); + // turn the OP_WRITE on + key.interestOps(key.interestOps() | OP_WRITE); - // turn the OP_WRITE on - key.interestOps(key.interestOps() | OP_WRITE); - - // in some cases it may not be possible to send out all the data before we - // fill out the write buffer, so this data will be send when the buffer - // is no longer full, and as such is treated as future work - if (futureWork != null) { - try { // we will complete what we can for now - boolean isComplete = futureWork.doWork(attached.entryWriter.in); - if (!isComplete) - attached.entryWriter.uncompletedWork = futureWork; - } catch (Exception e) { - LOG.error("", e); + // in some cases it may not be possible to send out all the data before + // we fill out the write buffer, so this data will be send when + // the buffer is no longer full, and as such is treated as future work + if (futureWork != null) { + try { // we will complete what we can for now + boolean isComplete = futureWork.doWork(attached.entryWriter.in); + if (!isComplete) + attached.entryWriter.uncompletedWork = futureWork; + } catch (Exception e) { + LOG.error("", e); + } } } - } - } else - externalizable.readExternalEntry(copies, segmentState, out); + } else + externalizable.readExternalEntry(copies, segmentState, out); - out.limit(limit); + out.limit(limit); - // skip onto the next entry - out.position(nextEntryPos); + // skip onto the next entry + out.position(nextEntryPos); - state = NOT_SET; - sizeInBytes = 0; + state = NOT_SET; + sizeInBytes = 0; + } + } finally { + if (LOG.isDebugEnabled()) + LOG.debug("Entries read: {}", entriesRead); } } From b179834ff16ce9b0e0127210e7414a123fc4bffa Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 19:03:08 +0000 Subject: [PATCH 09/24] added more under -"#### How many bytes is each entry" --- README.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/README.md b/README.md index c01a5bb47..df4420e4e 100644 --- a/README.md +++ b/README.md @@ -260,11 +260,13 @@ You set the maximum number of entries by the builder: ``` java ConcurrentMap map = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + constantValueSizeBySample("a long sample string"), .entries(1000) // set the max number of entries here .create(); ``` In this example above we have set 1000 entries. + We have optimised chronicle, So that you can have situations where you either don't use; - all the entries you have allowed for. This works best on Unix where the disk space and memory @@ -275,6 +277,19 @@ lines (128 bytes +), only the lines you touch sit in your CPU cache and if you h (8+ Kbytes) only the pages you touch use memory or disk. The CPU cache usage matters as it can be 10000x smaller than main memory. +#### How many bytes is each entry + +For example, if you had ant entry that had both Integer keys and values, these take exactly 4 bytes themselves. +There is also multi-map overhead per entry, ~6 bytes if your segment size is < 64k, + ~9 if < 16m and ~12 if > 16m entries per segment. If you create replicated map, + there is also 10 bytes per entry for replication purposes. + +We suggest you don't configure size for constant-sized keys or values, instead you can use the +builder methods .constantKeySizeBySample(sampleKey) and +.constantValueSizeBySample(sampleValue), For common types like Integer we suggest you don't use +these methods, for example ChronicleMap knows that Integer is 4 bytes long, Long is 8, etc. + + ### Size of space reserved on disk In linux, if you looked at the size of the 'file', it will report the used entry size so if you From 7ec3cdd53366bc025113e656ffe2c2092c1d9b3b Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 20:26:52 +0000 Subject: [PATCH 10/24] updated Entry size --- README.md | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index df4420e4e..aa1986f42 100644 --- a/README.md +++ b/README.md @@ -277,12 +277,15 @@ lines (128 bytes +), only the lines you touch sit in your CPU cache and if you h (8+ Kbytes) only the pages you touch use memory or disk. The CPU cache usage matters as it can be 10000x smaller than main memory. -#### How many bytes is each entry +#### Entry size -For example, if you had ant entry that had both Integer keys and values, these take exactly 4 bytes themselves. -There is also multi-map overhead per entry, ~6 bytes if your segment size is < 64k, - ~9 if < 16m and ~12 if > 16m entries per segment. If you create replicated map, - there is also 10 bytes per entry for replication purposes. +The size of each entry depends on the type of the Keys and Values, as some types are larger than others. For example, if an entry uses Integers for both the keys and values, both the key and value each take exactly 4 bytes. +There is also some other overhead which is internal to chronicle, such as its internal multi-map, which has an overhead of +* 6 bytes per entry for segment size of < 64k, +* 9 bytes per entry for segment size of < 16m but > 64k +* 12 bytes per entry for segment size of > 16m entries per segment. + +Also, if you create a replicated map, there is an additional 10 bytes per entry. We suggest you don't configure size for constant-sized keys or values, instead you can use the builder methods .constantKeySizeBySample(sampleKey) and From b659d8777b947482672b9116dad9e02b0086c514 Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 21:09:33 +0000 Subject: [PATCH 11/24] added performance test results --- README.md | 53 +++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 51 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index aa1986f42..07efdeb28 100644 --- a/README.md +++ b/README.md @@ -1126,13 +1126,13 @@ For in memory data-structures like a HashMap this isn’t a big problem. But in So if you don’t require the old value and don’t wish to block until your `put()` has been received by the server, then you may wish to consider using the following configuration : ``` java -.putReturnsNull +.putReturnsNull(true) ``` and also for the `remove()` method ``` java -.removeReturnsNull +.removeReturnsNull(true) ``` ``` java @@ -1153,6 +1153,55 @@ ChronicleMapBuilder.of(Integer.class, CharSequence.class) .create(); ``` +##### Performance + +The throughput and latency performance for different configurations. + +Tested using a test called BGChronicleTest. + +On one machine (i7 3.5 GHz) we have two persisted replicas run as + +-Dreplicas=2 eg.BGChronicleTest server + +On another machine (Dual Xeon 8 core 2.6 GHz) connected via a pair of Solarflare SFN5121T 10 Gig-E with onload enabled. + +-Dreplicas=2 -Dclients={see below} -DmaxRate=30000 -DreadRatio=2 eg.BGChronicleTest client + +2 clients +Throughput test +messages per seconds: 58,864 + +Latency test at 30,000 msg/sec +50% / 90% / 99% // 99.9% / 99.99% / worst latency was 33 / 80 / 111 // 120 / 148 / 3,921 us + +4 clients +Throughput test +messages per seconds: 94,006 + +Latency test at 30,000 msg/sec +50% / 90% / 99% // 99.9% / 99.99% / worst latency was 32 / 94 / 106 // 131 / 177 / 2,153 us + +8 clients +Throughput test +messages per seconds: 162,961 + +Latency test at 30,000 msg/sec +50% / 90% / 99% // 99.9% / 99.99% / worst latency was 35 / 94 / 117 // 140 / 167 / 1,685 us + +16 clients +Throughput test +messages per seconds: 267,097 + +Latency test at 30,000 msg/sec +50% / 90% / 99% // 99.9% / 99.99% / worst latency was 38 / 97 / 122 // 149 / 174 / 2,771 us + +24 clients +Throughput test +messages per seconds: 253,052 + +Latency test at 30,000 msg/sec +50% / 90% / 99% // 99.9% / 99.99% / worst latency was 40 / 99 / 121 // 151 / 243 / 3,669 us + ##### Close its always important to close `ChronicleMap`'s and `ChronicleSet` 's when you have finished with them From ed2df2895ca08c0d02a7ccdda24d55246af4157c Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 21:57:54 +0000 Subject: [PATCH 12/24] updated Entry size --- README.md | 29 ++++++++++++++--------------- 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 07efdeb28..a7a2a24bc 100644 --- a/README.md +++ b/README.md @@ -315,24 +315,23 @@ says the size is 71 MB after adding 10000 entries. You can see the size actually ### How Operating Systems differ -There is two key differences between Windows and Linux +As a pure java library, the same chronicle map java byte code can be run on Windows, Linux and Mac OSX. +However these operating systems work with memory mapped files differently, these differences effect how +chronicle is able to map memory to a file, and hence this can impact the total number of entries that you are +able to configure. -- Windows allocates memory and disk eagerly - Windows fails if you attempt to use more than 4 GB in a single mapping. This calculated as 4GB = 2^20 * 4 KB pages. This doesn't fail when you map the region, rather as you use it up. -It may be this limitation doesn't apply to newer or server based versions of Windows. -In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. -- Linux allocates memory and disk lazily. -- MacOSX allocates memory lazily and disk eagerly. +- Windows allocates memory and disk eagerly, Windows will fail if more than 4 GB is allocated in a single memory +mapping, ( calculated as 4GB = 2^20 * 4 KB pages). Windows doesn't fail when a memory mapped region is mapped, rather it will fail when it is used up. This limitation doesn't apply to newer or server based versions of Windows. Eager memory allocation means you can't map more than free memory, but it should reduce jitter when you use it. In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. +- Linux allocates memory and disk lazily. Linux systems see a performance degradation at around 200% of main memory. +- Mac OSX allocates memory lazily and disk eagerly. -On Windows, eager memory allocation means you can't map more than free memory, but it should reduce jitter when you use it. -Chronicle Map allocates head room which is a waste on Windows (Linux's sparse allocation means the head room has -little impact). +Chronicle Map allocates head room which is a waste on Windows (Linux's sparse allocation means the head room has little +impact). -Linux systems see a performance degradation at around 200% of main memory size. - -For production systems, we recommend; -- on Windows smaller map sizes of less than 4 GB each, and less than 50% main memory in total. -- on Linux small to large maps of less than double main memory. e.g. if you have a 128 GB server, have less than 256 GB of maps on the server. -- on MacOSX, we have no specific recommendations. +For production systems +- on Windows we recommend you use map sizes of less than 4 GB each, and less than 50% main memory in total. +- on Linux we recommend you use small to large maps of less than double main memory. e.g. if you have a 128 GB server, we recommend you have less than 256 GB of maps on the server. +- on Mac OSX, we have no specific recommendations. ### Chronicle Map Interface The Chronicle Map interface adds a few methods above an beyond the standard ConcurrentMap, From 60ad0c2ce9425a15b14fa671a307741b6b150459 Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 21:59:12 +0000 Subject: [PATCH 13/24] updated Entry size --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index a7a2a24bc..9bb5765a4 100644 --- a/README.md +++ b/README.md @@ -328,7 +328,7 @@ mapping, ( calculated as 4GB = 2^20 * 4 KB pages). Windows doesn't fail when a m Chronicle Map allocates head room which is a waste on Windows (Linux's sparse allocation means the head room has little impact). -For production systems +##### For production - on Windows we recommend you use map sizes of less than 4 GB each, and less than 50% main memory in total. - on Linux we recommend you use small to large maps of less than double main memory. e.g. if you have a 128 GB server, we recommend you have less than 256 GB of maps on the server. - on Mac OSX, we have no specific recommendations. From a2aaaa3246f260bea5ebdfff3688693fd8ff245e Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 22:05:40 +0000 Subject: [PATCH 14/24] change the example to use CharSequence --- README.md | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 9bb5765a4..34a51534c 100644 --- a/README.md +++ b/README.md @@ -350,18 +350,17 @@ supports the following methods : - [`ReadContext getUsingLocked(@NotNull K key, @NotNull V usingValue);`] (https://github.com/OpenHFT/Chronicle-Map#off-heap-storage-and-how-using-a-proxy-object-can-improve-performance) - [`WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue);`] (https://github.com/OpenHFT/Chronicle-Map#acquireusinglocked) -These methods let you provide the object which the data will be written to, even if the object is -immutable. For example +These methods let you provide the object which the data will be written to, but the value use to be mutable. For example ``` java -String myString = ""; -String myResult = map.getUsing("key", myString); +CharSequence using = new StringBuilder(); +CharSequence myResult = map.getUsing("key", using); // at this point the myString and myResult will both point to the same object ``` The `map.getUsing()` method is similar to `map.get()`, but because Chronicle Map stores its data off heap, if you were to call get("key"), a new object would be created each time, map.getUsing() works -by reusing the heap memory which was used by the original Object "myString". This technique provides +by reusing the heap memory which was used by the original Object "using". This technique provides you with better control over your object creation. Exactly like `map.getUsing()`, `map.acquireUsing()` will give you back a reference to an value From 3db6cac3a0ee9fac4d6fcf5be4a3bfbeb3816e43 Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 22:21:26 +0000 Subject: [PATCH 15/24] minor corrections --- README.md | 29 ++++++++++++++--------------- 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 34a51534c..1cd8a83f3 100644 --- a/README.md +++ b/README.md @@ -453,7 +453,7 @@ bond.getCoupon() lets say that it is only the `coupon` field that we are interested in, then its better not to have to deserialise the whole object that implements the `BondVOInterface`. The `ChronicleMapBuilder` will look a the types of keys and values that you use, If the value type is a simple accessor/mutator interface that is exposing a non nested pojo, which uses simple types - like String and primitives with corresponding get..() and + like `CharSequence` and primitives with corresponding get..() and set..() methods, Chronicle is able to generate off heap poxies so the whole object is not desrialized each time it is accessed, The off heap poxies are able to read and write into @@ -461,7 +461,7 @@ the off heap data structures directly, this reduced serialisation can give you a Below we show you how you can work directly with the off heap entries. ``` java - ChronicleMap chm = ChronicleMapBuilder + ChronicleMap chm = ChronicleMapBuilder .of(String.class, BondVOInterface.class) .keySize(10) .create(); @@ -470,7 +470,7 @@ Below we show you how you can work directly with the off heap entries. notice that the ``` java -.of(String.class, BondVOInterface.class) +.of(CharSequence.class, BondVOInterface.class) ``` value class, in our case `BondVOInterface.class` is an `interface` rather than a `class`, now @@ -542,14 +542,13 @@ context.close() // the lock will get released when this is called ``` #### acquireUsingLocked() - -just like getUsing(), acquireUsing() will also recycle the value you pass it, the following -code is a pattern that you will often come across, acquireUsing(key,value) offers this -functionality with a single method call : +Just like getUsing(), acquireUsing() will also recycle the value you pass it, the following +code is a pattern that you will often come across, `acquireUsing(key,value)` offers this +functionality i the example, below with a single method call : ``` java V acquireUsing(key,value) { - Lock l = ...; // the segement lock of the map + Lock l = ...; // the segment lock of the map l.lock(); try { V v = map.getUsing(key,value) @@ -564,12 +563,12 @@ V acquireUsing(key,value) { } ``` -if you are only accessing ChronicleMap from a single thread and you are not doing replication -and don't care about atomic reads, then its simpler ( and faster ) to use acquireUsing() otherwise we -recommend acquireUsingLocked(key,value) +If you are only accessing ChronicleMap from a single thread. If you are not doing replication +and don't care about atomic reads. Then its simpler ( and faster ) to use acquireUsing() otherwise we +recommend you use `acquireUsingLocked(key,value)` -because acquireUsing can end up creating an entry the acquireUsingLocked(key,value) method holds -a segment write lock, this is unlike the getUsing(key,using) method that holds a segment read lock. +Since the acquireUsing() method can end up creating an entry, the acquireUsingLocked(key,value) method must hold +a segment write lock, this is unlike getUsing(key,using) which only holds a segment read lock. ``` java BondVOInterface bond = ... // create your instance @@ -584,7 +583,7 @@ try (WriteContext context = map.acquireUsingLocked("one", bo } ``` -if after you have read the 'issueDate' and 'symbol' and you wish to remove the entry based on some +If after you have read the 'issueDate' and 'symbol' and you wish to remove the entry based on some business logic, it more efficient to use the 'context' to remove the entry, as the contents is already aware when the entry is in memory. @@ -593,7 +592,7 @@ already aware when the entry is in memory. It is possible for the size of your entry to be twice as large as the maximum entry size, we refer to this type of entry as an oversized entry. Oversized entries are there to cater for the case where only a small -percentage of your entries are twise as large as the others, in this case your large entry will +percentage of your entries are twice as large as the others, in this case your large entry will span across two entries. The alternative would be to increase your maximum entry size to be similar to the size of the largest entry, but this approach is wasteful of memory, especially when most entries are no where near the max entry size. From 831539133e8adc5f727fc30d22cdee63095505ea Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 22:23:01 +0000 Subject: [PATCH 16/24] minor corrections --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 1cd8a83f3..960a7357f 100644 --- a/README.md +++ b/README.md @@ -584,7 +584,7 @@ try (WriteContext context = map.acquireUsingLocked("one", bo ``` If after you have read the 'issueDate' and 'symbol' and you wish to remove the entry based on some -business logic, it more efficient to use the 'context' to remove the entry, as the contents is +business logic, its more efficient to use the 'context' to remove the entry, as the contents is already aware when the entry is in memory. ## Oversized Entries Support From 3d1cc76a055f2e50c3468fd25656a4f94f1c14d4 Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 22:25:45 +0000 Subject: [PATCH 17/24] updated oversized --- README.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 960a7357f..e2fe20761 100644 --- a/README.md +++ b/README.md @@ -589,11 +589,11 @@ already aware when the entry is in memory. ## Oversized Entries Support -It is possible for the size of your entry to be twice as large as the maximum entry size, +It is possible for the size of your entry to be 64 times as large as the maximum entry size, we refer to this type of entry as an oversized entry. Oversized entries are there to cater for the case where only a small -percentage of your entries are twice as large as the others, in this case your large entry will -span across two entries. The alternative would be to increase your maximum entry size to be similar +percentage of your entries are larger than the others, in this case your large entry will +span across a number of entries. The alternative would be to increase your maximum entry size to be similar to the size of the largest entry, but this approach is wasteful of memory, especially when most entries are no where near the max entry size. From f3f3c84c0f1332d06044b52015d29d8d97fbc260 Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 22:28:38 +0000 Subject: [PATCH 18/24] updated import --- README.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index e2fe20761..cfa44ebee 100644 --- a/README.md +++ b/README.md @@ -628,10 +628,12 @@ void putAll(File fromFile) throws IOException; Its only the entries of your map that are exported, not the configuration of your map. So care must be taken to populate the data in to a map of the correct Key/Value type and with enough available entries. When importing data : + * entries that are in the map but not in the JSON file will remain untouched. * entries that are in the map and in the JSON file will be updated. -* entries that are not in the map but are in the JSON file wil added. -* In other words importing data into a Chronicle Map works like `map.putAll()`. +* entries that are not in the map but are in the JSON file will be added. + +In other words importing data into a Chronicle Map works like `map.putAll()`. When Importing data if you are also writing to the map at the same time, the last update will win. In other words a write lock is not held for the entire import process. From dba44661a03e7b70601694002856e1b9d3710b71 Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 22:32:59 +0000 Subject: [PATCH 19/24] correct comment on Close() --- README.md | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/README.md b/README.md index cfa44ebee..0c88baf3b 100644 --- a/README.md +++ b/README.md @@ -652,10 +652,7 @@ Its recommended that you call close() once you have finished working with a Chro map.close() ``` -You only need to close to clean up resources deterministically. If your program is exiting, -you don't need to close the collection, as Chronicle never knows when the program might crash, -so we have designed it so you don't have to close() it. - +When using replication its important to call close to clean up the NIO socket connections. Failure to call close may prevent you from restarting a replicated map on the same port. WARNING : If you call close too early before you have finished working with the map, this can cause your JVM to crash. Close MUST BE the last thing that you do with the map. From 78ca1c55fd0dcdcc499544d65ebf1f840dd6438f Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Sun, 7 Dec 2014 22:35:10 +0000 Subject: [PATCH 20/24] correct comment on Close() --- README.md | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 0c88baf3b..61895d93c 100644 --- a/README.md +++ b/README.md @@ -652,10 +652,13 @@ Its recommended that you call close() once you have finished working with a Chro map.close() ``` -When using replication its important to call close to clean up the NIO socket connections. Failure to call close may prevent you from restarting a replicated map on the same port. -WARNING : If you call close too early before you have finished working with the map, this can cause -your JVM to crash. Close MUST BE the last thing that you do with the map. +When using replication its important to call close to clean up the TCP/IP socket connection. +Failure to call close may prevent you from restarting a replicated map on the same port. +###### WARNING + +If you call close too early before you have finished working with the map, this can cause +your JVM to crash. Close MUST BE the last thing that you do with the map. # TCP / UDP Replication From c7bfb46b0b3d1340c629bdee1e1374c603303eac Mon Sep 17 00:00:00 2001 From: boundedbuffer Date: Mon, 8 Dec 2014 09:27:37 +0000 Subject: [PATCH 21/24] correct comment on Close() --- README.md | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 61895d93c..9e26588df 100644 --- a/README.md +++ b/README.md @@ -648,12 +648,16 @@ VALUES, future versions will support a binary encoding of objects that are `net. ## Close Unlike ConcurrentHashMap, chronicle map stores its data off heap, often in a memory mapped file. Its recommended that you call close() once you have finished working with a Chronicle Map. + ``` java map.close() ``` -When using replication its important to call close to clean up the TCP/IP socket connection. -Failure to call close may prevent you from restarting a replicated map on the same port. +This is especially important when working with chronicle map replication, as failure to call close may prevent +you from restarting a replicated map on the same port. In the event that your application crashes it may not +be possible to call close(). Your operating system will usually close dangling ports automatically, +so although it is recommended that you close() when you have finished with the map, +its not something that you must do, its just something that we recommend you should do. ###### WARNING From b659c56d3485854abce7573b2b633cac15c4393f Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 8 Dec 2014 10:12:19 +0000 Subject: [PATCH 22/24] HCOLL-230 - created nextEntry() --- .../chronicle/map/StatelessChronicleMap.java | 83 ++++++++++--------- .../net/openhft/chronicle/map/Builder.java | 2 +- .../chronicle/map/TcpReplicationSoakTest.java | 21 +++-- 3 files changed, 55 insertions(+), 51 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 6ee6aaa18..d56d34a17 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -992,9 +992,49 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, InterruptedException { - int remainingBytes = 0; + + int remainingBytes = nextEntry(timeoutTime, transactionId); + + if (inBytes.capacity() < remainingBytes) { + long pos = inBytes.position(); + long limit = inBytes.position(); + inBytes.position(limit); + resizeBufferInBuffer(remainingBytes, pos); + } else + inBytes.limit(inBytes.capacity()); + + // block until we have received all the bytes in this chunk + receive(remainingBytes, timeoutTime); + + final boolean isException = inBytes.readBoolean(); + + if (isException) { + Throwable throwable = (Throwable) inBytes.readObject(); + try { + Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); + stackTrace.setAccessible(true); + List stes = new ArrayList<>(Arrays.asList((StackTraceElement[]) stackTrace.get(throwable))); + // prune the end of the stack. + for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) { + stes.remove(i); + } + InetSocketAddress address = config.remoteAddress(); + stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort())); + StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace(); + for (int i = 4; i < stackTrace2.length; i++) + stes.add(stackTrace2[i]); + stackTrace.set(throwable, stes.toArray(new StackTraceElement[stes.size()])); + } catch (Exception ignore) { + } + NativeBytes.UNSAFE.throwException(throwable); + } + return inBytes; + } + + private int nextEntry(long timeoutTime, long transactionId) throws IOException, InterruptedException { + int remainingBytes; for (; ; ) { // read the next item from the socket @@ -1062,46 +1102,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon pause(); } - - - final int minBufferSize = remainingBytes; - - if (inBytes.capacity() < minBufferSize) { - long pos = inBytes.position(); - long limit = inBytes.position(); - inBytes.position(limit); - resizeBufferInBuffer(minBufferSize, pos); - } else - inBytes.limit(inBytes.capacity()); - - // block until we have received all the bytes in this chunk - receive(remainingBytes, timeoutTime); - - final boolean isException = inBytes.readBoolean(); - - if (isException) { - Throwable throwable = (Throwable) inBytes.readObject(); - try { - Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); - stackTrace.setAccessible(true); - List stes = new ArrayList<>(Arrays.asList((StackTraceElement[]) stackTrace.get(throwable))); - // prune the end of the stack. - for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) { - stes.remove(i); - } - InetSocketAddress address = config.remoteAddress(); - stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort())); - StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace(); - for (int i = 4; i < stackTrace2.length; i++) - stes.add(stackTrace2[i]); - stackTrace.set(throwable, stes.toArray(new StackTraceElement[stes.size()])); - } catch (Exception ignore) { - } - NativeBytes.UNSAFE.throwException(throwable); - } - - - return inBytes; + return remainingBytes; } private void clearParked() { diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java index 923c0a800..79d4ffc3c 100644 --- a/src/test/java/net/openhft/chronicle/map/Builder.java +++ b/src/test/java/net/openhft/chronicle/map/Builder.java @@ -32,7 +32,7 @@ import java.util.concurrent.TimeUnit; */ public class Builder { - public static final int SIZE = 10000; + public static final int SIZE = 1000; // added to ensure uniqueness static int count; static String WIN_OS = "WINDOWS"; diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index 14dd1dcfe..3d17c22d3 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -5,7 +5,10 @@ import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.model.Byteable; import net.openhft.lang.model.DataValueClasses; import net.openhft.lang.values.IntValue; -import org.junit.*; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; import java.io.Closeable; import java.io.IOException; @@ -100,19 +103,19 @@ public class TcpReplicationSoakTest { final int select = rnd.nextInt(2); final ChronicleMap map = select > 0 ? map1 : map2; - if (rnd.nextBoolean()) { - map.put(rnd.nextInt(Builder.SIZE), "test" + j); - } else { - map.remove(rnd.nextInt(Builder.SIZE)); + if (rnd.nextBoolean()) { + map.put(rnd.nextInt(Builder.SIZE), "test" + j); + } else { + map.remove(rnd.nextInt(Builder.SIZE)); + } } } - } - Thread.sleep(1000); + Thread.sleep(1000); System.out.println("\nwaiting till equal"); - waitTillEqual(10000); + waitTillEqual(10000); - Assert.assertEquals(new TreeMap(map1), new TreeMap(map2)); + Assert.assertEquals(map1, map2); } From 9065044ef49f70ebccab1f9e5d08b13724cb10bc Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 8 Dec 2014 10:14:47 +0000 Subject: [PATCH 23/24] HCOLL-230 - created nextEntry() --- .../chronicle/map/StatelessChronicleMap.java | 101 +++++++++--------- 1 file changed, 52 insertions(+), 49 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 29fc940f8..3f470699f 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -40,7 +40,9 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.nio.channels.SocketChannel; import java.util.*; -import java.util.concurrent.*; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; @@ -204,9 +206,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon doHandShaking(result); break; } catch (IOException e) { - if (closeables!=null) closeables.closeQuietly(); + if (closeables != null) closeables.closeQuietly(); } catch (Exception e) { - if (closeables!=null) closeables.closeQuietly(); + if (closeables != null) closeables.closeQuietly(); throw e; } } @@ -237,7 +239,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon clientChannel.connect(remoteAddress); doHandShaking(clientChannel); } catch (IOException e) { - if (closeables!=null) closeables.closeQuietly(); + if (closeables != null) closeables.closeQuietly(); } } @@ -341,7 +343,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @SuppressWarnings("NullableProblems") - public boolean remove( Object key, Object value) { + public boolean remove(Object key, Object value) { if (key == null) throw new NullPointerException(); @@ -360,7 +362,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } @SuppressWarnings("NullableProblems") - public V replace(K key, V value) { + public V replace(K key, V value) { if (key == null || value == null) throw new NullPointerException(); @@ -973,16 +975,57 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - // this is a transactionid and size that has been read by another thread. + // this is a transaction id and size that has been read by another thread. private volatile long parkedTransactionId; private volatile int parkedRemainingBytes; private volatile long parkedTransactionTimeStamp; private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, InterruptedException { - int remainingBytes = 0; + + int remainingBytes = nextEntry(timeoutTime, transactionId); + + if (inBytes.capacity() < remainingBytes) { + long pos = inBytes.position(); + long limit = inBytes.position(); + inBytes.position(limit); + resizeBufferInBuffer(remainingBytes, pos); + } else + inBytes.limit(inBytes.capacity()); + + // block until we have received all the bytes in this chunk + receive(remainingBytes, timeoutTime); + + final boolean isException = inBytes.readBoolean(); + + if (isException) { + Throwable throwable = (Throwable) inBytes.readObject(); + try { + Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); + stackTrace.setAccessible(true); + List stes = new ArrayList<>(Arrays.asList((StackTraceElement[]) stackTrace.get(throwable))); + // prune the end of the stack. + for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) { + stes.remove(i); + } + InetSocketAddress address = config.remoteAddress(); + stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort())); + StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace(); + //noinspection ManualArrayToCollectionCopy + for (int i = 4; i < stackTrace2.length; i++) + stes.add(stackTrace2[i]); + stackTrace.set(throwable, stes.toArray(new StackTraceElement[stes.size()])); + } catch (Exception ignore) { + } + NativeBytes.UNSAFE.throwException(throwable); + } + return inBytes; + } + + private int nextEntry(long timeoutTime, long transactionId) throws IOException { + int remainingBytes; for (; ; ) { // read the next item from the socket @@ -1050,47 +1093,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon pause(); } - - - final int minBufferSize = remainingBytes; - - if (inBytes.capacity() < minBufferSize) { - long pos = inBytes.position(); - long limit = inBytes.position(); - inBytes.position(limit); - resizeBufferInBuffer(minBufferSize, pos); - } else - inBytes.limit(inBytes.capacity()); - - // block until we have received all the bytes in this chunk - receive(remainingBytes, timeoutTime); - - final boolean isException = inBytes.readBoolean(); - - if (isException) { - Throwable throwable = (Throwable) inBytes.readObject(); - try { - Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); - stackTrace.setAccessible(true); - List stes = new ArrayList<>(Arrays.asList((StackTraceElement[]) stackTrace.get(throwable))); - // prune the end of the stack. - for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) { - stes.remove(i); - } - InetSocketAddress address = config.remoteAddress(); - stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort())); - StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace(); - //noinspection ManualArrayToCollectionCopy - for (int i = 4; i < stackTrace2.length; i++) - stes.add(stackTrace2[i]); - stackTrace.set(throwable, stes.toArray(new StackTraceElement[stes.size()])); - } catch (Exception ignore) { - } - NativeBytes.UNSAFE.throwException(throwable); - } - - - return inBytes; + return remainingBytes; } private void clearParked() { From 7486723d8ce45d22e2dab7bb9725e3880063d5fe Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 8 Dec 2014 12:35:08 +0000 Subject: [PATCH 24/24] HCOLL-230 - updated the protocol between the client and the server to include the id and timestamp --- .../map/AbstractChannelReplicator.java | 2 +- .../chronicle/map/StatelessChronicleMap.java | 96 +++++-------------- .../openhft/chronicle/map/TcpReplicator.java | 87 +++++++++-------- .../chronicle/map/StatelessClientTest.java | 3 +- 4 files changed, 74 insertions(+), 114 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 80267adbf..8f87caed2 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -53,7 +53,7 @@ abstract class AbstractChannelReplicator implements Closeable { public static final int SIZE_OF_SIZE = 4; public static final int SIZE_OF_TRANSACTION_ID = 8; - public static final int SIZE_OF_IS_EXCEPTION = 1; + public static final int SIZE_OF_TIME_SHIFT = 2; private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class); diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 3f470699f..5a7a82458 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -47,8 +47,7 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; import static java.nio.ByteBuffer.allocateDirect; -import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; -import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTION_ID; +import static net.openhft.chronicle.map.AbstractChannelReplicator.*; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.*; /** @@ -127,6 +126,15 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon UPDATE_FOR_KEY } + + // used by the enterprise version + private int identifier; + + void identifier(int identifier) { + this.identifier = identifier; + } + + @NotNull private final AtomicLong transactionID = new AtomicLong(0); @@ -489,56 +497,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private synchronized void writeEntriesForPutAll(@NotNull Map map) { - final int numberOfEntries = map.size(); - int numberOfEntriesReadSoFar = 0; - - outBytes.writeStopBit(numberOfEntries); - assert outBytes.limit() == outBytes.capacity(); - - ThreadLocalCopies copies = keyWriterWithSize.getCopies(null); - final Object keyWriter = keyWriterWithSize.writerForLoop(copies); - copies = valueWriterWithSize.getCopies(copies); - final Object valueWriter = valueWriterWithSize.writerForLoop(copies); - - for (final Map.Entry e : map.entrySet()) { - final K key = e.getKey(); - final V value = e.getValue(); - if (key == null || value == null) - throw new NullPointerException(); - - numberOfEntriesReadSoFar++; - - long start = outBytes.position(); - - // putAll if a bit more complicated than the others - // as the volume of data could cause us to have to resize our buffers - resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar, start); - - final Class keyClass = key.getClass(); - if (!kClass.isAssignableFrom(keyClass)) { - throw new ClassCastException("key=" + key + " is of type=" + keyClass + " " + - "and should" + - " be of type=" + kClass); - } - - writeKeyInLoop(key, keyWriter, copies); - - final Class valueClass = value.getClass(); - if (!vClass.isAssignableFrom(valueClass)) - throw new ClassCastException("value=" + value + " is of type=" + valueClass + - " and " + - "should be of type=" + vClass); - - writeValueInLoop(value, valueWriter, copies); - - final int len = (int) (outBytes.position() - start); - - if (len > maxEntrySize) - maxEntrySize = len; - } - } - private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) { final long remaining = outBytes.remaining(); @@ -723,7 +681,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public void putAll(@NotNull Map map) { - final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : + final long sizeLocation = putReturnsNull ? writeEventAnSkip(PUT_ALL_WITHOUT_ACC) : writeEventAnSkip(PUT_ALL); final long startTime = System.currentTimeMillis(); @@ -856,6 +814,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // skips for the transaction id outBytes.skip(SIZE_OF_TRANSACTION_ID); + outBytes.skip(SIZE_OF_TIME_SHIFT); + outBytes.writeByte(identifier); return sizeLocation; } @@ -934,10 +894,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon if (LOG.isDebugEnabled()) LOG.debug("sending data with transactionId=" + transactionId); - writeSizeAndTransactionIdAt(sizeLocation, transactionId); + writeSizeAndTransactionIdAt(sizeLocation, transactionId, startTime); // send out all the bytes - send(outBytes, timeoutTime); + writeBytesToSocket(outBytes, timeoutTime); outBytes.clear(); outBytes.buffer().clear(); break; @@ -1141,7 +1101,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private void send(@NotNull final Bytes out, long timeoutTime) throws IOException { + private void writeBytesToSocket(@NotNull final Bytes out, long timeoutTime) throws IOException { outBuffer.limit((int) out.position()); outBuffer.position(0); @@ -1160,7 +1120,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon throw new RemoteCallTimeoutException(); } - private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId) { + private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId, long startTime) { final long size = outBytes.position() - locationOfSize; final long pos = outBytes.position(); outBytes.position(locationOfSize); @@ -1168,15 +1128,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon assert size0 > 0; outBytes.writeInt(size0); + assert transactionId != 0; outBytes.writeLong(transactionId); - outBytes.position(pos); - } + outBytes.writeUnsignedShort((int) (transactionId - startTime)); - private void writeSizeAt(long locationOfSize) { - final long size = outBytes.position() - locationOfSize; - int size0 = (int) size - SIZE_OF_SIZE; - assert size0 > 0; - outBytes.writeInt(locationOfSize, size0); + outBytes.position(pos); } private long markSizeLocation() { @@ -1545,9 +1501,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = eventReturnsNull(eventId) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); + final long sizeLocation = writeEventAnSkip(eventId); copies = writeKey(key); copies = writeValue(value, copies); @@ -1577,9 +1531,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = eventReturnsNull(eventId) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); + final long sizeLocation = writeEventAnSkip(eventId); copies = writeKey(key); copies = writeValue(value1, copies); @@ -1611,9 +1563,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = eventReturnsNull(eventId) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); + final long sizeLocation = writeEventAnSkip(eventId); copies = writeKey(key); transactionId = send(sizeLocation, startTime); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 50bafefc5..958d42ee1 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1329,28 +1329,39 @@ class StatelessServerConnector { @NotNull final ByteBufferBytes reader) { final StatelessChronicleMap.EventId event = VALUES[eventId]; - // these methods don't return a result + long transactionId = reader.readLong(); + + // the time stamp and the transaction are usually the same, or out by the shift + int timestampShift = reader.readUnsignedShort(); + long timestamp = transactionId - timestampShift; + + byte identifier = reader.readByte(); + + + // these methods don't return a result to the client or don't return a result to the + // client immediately switch (event) { case KEY_SET: - return keySet(reader, writer); + return keySet(reader, writer, transactionId); case VALUES: - return values(reader, writer); + return values(reader, writer, transactionId); case ENTRY_SET: - return entrySet(reader, writer); + return entrySet(reader, writer, transactionId); case PUT_WITHOUT_ACC: - return put(reader); + return put(reader, timestamp, identifier); case PUT_ALL_WITHOUT_ACC: - return putAll(reader); + return putAll(reader, timestamp, identifier); case REMOVE_WITHOUT_ACC: - return remove(reader); + return remove(reader, timestamp, identifier); } - final long sizeLocation = reflectTransactionId(reader, writer); + final long sizeLocation = reflectTransactionId(writer, transactionId); + // these methods return a result @@ -1368,34 +1379,33 @@ class StatelessServerConnector { return containsValue(reader, writer, sizeLocation); case GET: - return get(reader, writer, sizeLocation); + return get(reader, writer, sizeLocation, timestamp); case PUT: - return put(reader, writer, sizeLocation); + return put(reader, writer, sizeLocation, timestamp, identifier); case REMOVE: - return remove(reader, writer, sizeLocation); + return remove(reader, writer, sizeLocation, timestamp, identifier); case CLEAR: - return clear(writer, sizeLocation); + return clear(writer, sizeLocation, timestamp, identifier); case REPLACE: - return replace(reader, writer, sizeLocation); + return replace(reader, writer, sizeLocation, timestamp, identifier); case REPLACE_WITH_OLD_AND_NEW_VALUE: - return replaceWithOldAndNew(reader, writer, sizeLocation); + return replaceWithOldAndNew(reader, writer, sizeLocation, timestamp, identifier); case PUT_IF_ABSENT: - return putIfAbsent(reader, writer, sizeLocation); + return putIfAbsent(reader, writer, sizeLocation, timestamp, identifier); case REMOVE_WITH_VALUE: - return removeWithValue(reader, writer, sizeLocation); - + return removeWithValue(reader, writer, sizeLocation, timestamp, identifier); case TO_STRING: return toString(writer, sizeLocation); case PUT_ALL: - return putAll(reader, writer, sizeLocation); + return putAll(reader, writer, sizeLocation, timestamp, identifier); case HASH_CODE: return hashCode(writer, sizeLocation); @@ -1444,7 +1454,7 @@ class StatelessServerConnector { } @Nullable - private Work removeWithValue(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work removeWithValue(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { writer.writeBoolean(map.removeWithValue(reader)); @@ -1463,7 +1473,8 @@ class StatelessServerConnector { } @Nullable - private Work replaceWithOldAndNew(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work replaceWithOldAndNew(Bytes reader, @NotNull Bytes writer, final long + sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.replaceWithOldAndNew(reader, writer); @@ -1585,7 +1596,9 @@ class StatelessServerConnector { } @Nullable - private Work get(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work get(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long + transactionId) { + try { if (MAP_SUPPORTS_BYTES) { map.get(reader, writer); @@ -1603,7 +1616,7 @@ class StatelessServerConnector { @SuppressWarnings("SameReturnValue") @Nullable - private Work put(Bytes reader) { + private Work put(Bytes reader, long timestamp, byte id) { if (MAP_SUPPORTS_BYTES) { map.put(reader); } else { @@ -1618,7 +1631,7 @@ class StatelessServerConnector { } @Nullable - private Work put(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work put(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.put(reader, writer); @@ -1638,7 +1651,7 @@ class StatelessServerConnector { @SuppressWarnings("SameReturnValue") @Nullable - private Work remove(Bytes reader) { + private Work remove(Bytes reader, long timestamp, byte id) { if (MAP_SUPPORTS_BYTES) { map.removeKeyAsBytes(reader); } else { @@ -1650,7 +1663,7 @@ class StatelessServerConnector { } @Nullable - private Work remove(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work remove(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.remove(reader, writer); @@ -1669,7 +1682,7 @@ class StatelessServerConnector { } @Nullable - private Work putAll(@NotNull Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work putAll(@NotNull Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.putAll(reader); @@ -1686,7 +1699,7 @@ class StatelessServerConnector { @SuppressWarnings("SameReturnValue") @Nullable - private Work putAll(@NotNull Bytes reader) { + private Work putAll(@NotNull Bytes reader, long timestamp, byte id) { if (MAP_SUPPORTS_BYTES) { map.putAll(reader); } else { @@ -1697,7 +1710,7 @@ class StatelessServerConnector { } @Nullable - private Work clear(@NotNull Bytes writer, final long sizeLocation) { + private Work clear(@NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { map.clear(); } catch (Throwable e) { @@ -1709,8 +1722,7 @@ class StatelessServerConnector { } @Nullable - private Work values(@NotNull Bytes reader, @NotNull Bytes writer) { - final long transactionId = reader.readLong(); + private Work values(@NotNull Bytes reader, @NotNull Bytes writer, final long transactionId) { Collection values; @@ -1752,8 +1764,7 @@ class StatelessServerConnector { } @Nullable - private Work keySet(@NotNull Bytes reader, @NotNull final Bytes writer) { - final long transactionId = reader.readLong(); + private Work keySet(@NotNull Bytes reader, @NotNull final Bytes writer, final long transactionId) { Set ks; @@ -1795,8 +1806,7 @@ class StatelessServerConnector { } @Nullable - private Work entrySet(@NotNull final Bytes reader, @NotNull Bytes writer) { - final long transactionId = reader.readLong(); + private Work entrySet(@NotNull final Bytes reader, @NotNull Bytes writer, final long transactionId) { final Set> entries; @@ -1844,7 +1854,7 @@ class StatelessServerConnector { } @Nullable - private Work putIfAbsent(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work putIfAbsent(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.putIfAbsent(reader, writer); @@ -1863,7 +1873,7 @@ class StatelessServerConnector { } @Nullable - private Work replace(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work replace(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.replaceKV(reader, writer); @@ -1881,8 +1891,7 @@ class StatelessServerConnector { return null; } - private long reflectTransactionId(@NotNull Bytes reader, @NotNull Bytes writer) { - final long transactionId = reader.readLong(); + private long reflectTransactionId(@NotNull Bytes writer, final long transactionId) { final long sizeLocation = writer.position(); writer.skip(SIZE_OF_SIZE); assert transactionId != 0; @@ -1940,7 +1949,7 @@ class StatelessServerConnector { @Nullable private Work sendException(@NotNull Bytes reader, @NotNull Bytes writer, @NotNull Throwable e) { - final long sizeLocation = reflectTransactionId(reader, writer); + final long sizeLocation = reflectTransactionId(writer, reader.readLong()); return sendException(writer, sizeLocation, e); } diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 1d4288b61..c502b49ed 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -157,8 +157,9 @@ public class StatelessClientTest { } statelessMap.putAll(payload); + assertEquals(SIZE, serverMap.size()); - Set> entries = statelessMap.entrySet(); + Set> entries = statelessMap.entrySet(); Map.Entry next = entries.iterator().next(); assertEquals("some value=" + next.getKey(), next.getValue());