From e951a4339acdbd52ba34b371e83bf79ff28ab804 Mon Sep 17 00:00:00 2001 From: leventov Date: Fri, 19 Dec 2014 23:33:10 +0700 Subject: [PATCH 1/7] Reduced unnecessary and wrong ChronicleMapBuilder.entrySize() usage --- .../java/eg/TestNesterInterfaceMarshal.java | 43 ++++++---- .../chronicle/map/CHMLatencyTestMain.java | 1 - .../chronicle/map/ChannelReplicationTest.java | 8 +- .../chronicle/map/ChronicleMapTest.java | 82 +++++++++++-------- .../chronicle/map/EntryCountMapTest.java | 39 ++++++--- .../openhft/chronicle/map/MapByNameTest.java | 2 +- .../openhft/chronicle/map/OSResizesMain.java | 10 +-- .../chronicle/map/StatelessClientTest.java | 12 ++- .../map/TcpReplicationLargeEntryTest.java | 11 ++- ...ReplicationWithAcquireUsingLockedTest.java | 20 +++-- .../chronicle/map/example/ServersMapMain.java | 15 ++-- .../chronicle/map/example/StringsMapMain.java | 13 ++- .../pingpong_latency/PingPongCASLeft.java | 5 +- .../map/utility/ProcessInstanceLimiter.java | 2 +- 14 files changed, 160 insertions(+), 103 deletions(-) diff --git a/src/test/java/eg/TestNesterInterfaceMarshal.java b/src/test/java/eg/TestNesterInterfaceMarshal.java index 536573559..e03d8d69f 100644 --- a/src/test/java/eg/TestNesterInterfaceMarshal.java +++ b/src/test/java/eg/TestNesterInterfaceMarshal.java @@ -29,10 +29,13 @@ public class TestNesterInterfaceMarshal { public void testReplicatedMap() throws Exception { ChronicleMap chronicleMap = null; - TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS); + TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = + TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS); try { - ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class) - .entries(5000L).entrySize(2000) + ChronicleMapBuilder builder = ChronicleMapBuilder + .of(CharSequence.class, TestInstrumentVOInterface.class) + .entries(5000L) + .keySize(10) .replication((byte) 1, tcpTransportAndNetworkConfig); @@ -44,20 +47,24 @@ public class TestNesterInterfaceMarshal { ChronicleMap map = chronicleMap; ChronicleMap chronicleMap1 = null; - TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig1 = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)).heartBeatInterval(1L, TimeUnit.SECONDS); + TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig1 = + TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)) + .heartBeatInterval(1L, TimeUnit.SECONDS); try { - ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class) + ChronicleMapBuilder builder = ChronicleMapBuilder + .of(CharSequence.class, TestInstrumentVOInterface.class) .putReturnsNull(true) .removeReturnsNull(true) - .entries(5000L).entrySize(2000) + .entries(5000L).keySize(10) .replication((byte) 2, tcpTransportAndNetworkConfig1); chronicleMap1 = builder.create(); } catch (Exception e) { - System.out.println("*********************Error(s) creating launcher instrument cache: " + e); + System.out.println("*********************Error(s) creating launcher " + + "instrument cache: " + e); } ChronicleMap replicatedMap = chronicleMap1; @@ -107,10 +114,13 @@ public class TestNesterInterfaceMarshal { protected ChronicleMap createReplicatedMap1() { ChronicleMap chronicleMap = null; - TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS); + TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = + TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS); try { - ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class) - .entries(5000L).entrySize(2000) + ChronicleMapBuilder builder = ChronicleMapBuilder + .of(CharSequence.class, TestInstrumentVOInterface.class) + .entries(5000L) + .keySize(10) .replication((byte) 1, tcpTransportAndNetworkConfig); @@ -125,20 +135,25 @@ public class TestNesterInterfaceMarshal { protected ChronicleMap createReplicatedMap2() { ChronicleMap chronicleMap = null; - TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)).heartBeatInterval(1L, TimeUnit.SECONDS); + TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = + TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)) + .heartBeatInterval(1L, TimeUnit.SECONDS); try { - ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class) + ChronicleMapBuilder builder = ChronicleMapBuilder + .of(CharSequence.class, TestInstrumentVOInterface.class) .putReturnsNull(true) .removeReturnsNull(true) - .entries(5000L).entrySize(2000) + .entries(5000L) + .keySize(10) .replication((byte) 2, tcpTransportAndNetworkConfig); chronicleMap = builder.create(); } catch (Exception e) { - System.out.println("*********************Error(s) creating launcher instrument cache: " + e); + System.out.println("*********************Error(s) creating launcher " + + "instrument cache: " + e); } return chronicleMap; } diff --git a/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java b/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java index 8e6863439..980155837 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java +++ b/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java @@ -78,7 +78,6 @@ public class CHMLatencyTestMain { ChronicleMap countersMap = ChronicleMapBuilder.of(LongValue.class, LongValue.class) .entries(KEYS) - .entrySize(16) .createPersistedTo(file); // add keys diff --git a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java index cc1d5db98..f82596b37 100644 --- a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java @@ -62,7 +62,7 @@ public class ChannelReplicationTest { map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(2) - .entrySize(100) + .constantValueSizeBySample("EXAMPLE-1") .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); } @@ -76,7 +76,7 @@ public class ChannelReplicationTest { map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(2) - .entrySize(100) + .constantValueSizeBySample("EXAMPLE-1") .instance().replicatedViaChannel(hubB.createChannel((short) 1)).create(); } } @@ -111,12 +111,12 @@ public class ChannelReplicationTest { map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(2) - .entrySize(100) + .constantValueSizeBySample("EXAMPLE-1") .instance().replicatedViaChannel(hubB.createChannel((short) 2)).create(); map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(2) - .entrySize(100) + .constantValueSizeBySample("EXAMPLE-1") .instance().replicatedViaChannel(hubA.createChannel((short) 2)).create(); map2a.put(1, "EXAMPLE-2"); diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index 7f085d7a2..adf4b7049 100755 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -103,17 +103,12 @@ public class ChronicleMapTest { } private static ChronicleMap getSharedMap( - long entries, int segments, int entrySize) throws IOException { - return getSharedMap(entries, segments, entrySize, OF_4_BYTES); - } - - private static ChronicleMap getSharedMap( - long entries, int segments, int entrySize, Alignment alignment) + long entries, int segments, Alignment alignment, int keySize) throws IOException { return ChronicleMapBuilder.of(CharSequence.class, LongValue.class) .entries(entries) .minSegments(segments) - .entrySize(entrySize) + .keySize(keySize) .entryAndValueAlignment(alignment) .create(); } @@ -204,7 +199,7 @@ public class ChronicleMapTest { int count = 300; final ChronicleMap map = ChronicleMapBuilder.of(Object.class, Object.class) - .entrySize(count) + .entries(count) .minSegments(2).create(); for (int i = 1; i < count; i++) { @@ -414,10 +409,13 @@ public class ChronicleMapTest { @Test public void testAcquireWithNullContainer() throws Exception { - ChronicleMap map = getSharedMap(10 * 1000, 128, 24); + ChronicleMap map = + ChronicleMapBuilder.of(CharSequence.class, LongValue.class) + .entries(1000) + .entryAndValueAlignment(OF_4_BYTES) + .create(); map.acquireUsing("key", new LongValue$$Native()); assertEquals(0, map.acquireUsing("key", null).getValue()); - map.close(); } @@ -454,7 +452,11 @@ public class ChronicleMapTest { @Test public void testGetWithNullContainer() throws Exception { - ChronicleMap map = getSharedMap(10 * 1000, 128, 24); + ChronicleMap map = + ChronicleMapBuilder.of(CharSequence.class, LongValue.class) + .entries(10) + .entryAndValueAlignment(OF_4_BYTES) + .create(); map.acquireUsing("key", new LongValue$$Native()); assertEquals(0, map.getUsing("key", null).getValue()); @@ -463,7 +465,11 @@ public class ChronicleMapTest { @Test public void testGetWithoutAcquireFirst() throws Exception { - ChronicleMap map = getSharedMap(10, 1, 24); + ChronicleMap map = + ChronicleMapBuilder.of(CharSequence.class, LongValue.class) + .entries(10) + .entryAndValueAlignment(OF_4_BYTES) + .create(); assertNull(map.getUsing("key", DataValueClasses.newDirectReference(LongValue.class))); map.close(); @@ -485,9 +491,9 @@ public class ChronicleMapTest { public void testAcquireAndGet() throws IOException, ClassNotFoundException, IllegalAccessException, InstantiationException { int entries = 3/*00 * 1000*/; - testAcquireAndGet(getSharedMap(entries, 1, 24, OF_4_BYTES), entries); - testAcquireAndGet(getSharedMap(entries, 1, 24, NO_ALIGNMENT), entries); - testAcquireAndGet(getSharedMap(entries, 1, 24, OF_8_BYTES), entries); + testAcquireAndGet(getSharedMap(entries, 1, OF_4_BYTES, 10), entries); + testAcquireAndGet(getSharedMap(entries, 1, NO_ALIGNMENT, 10), entries); + testAcquireAndGet(getSharedMap(entries, 1, OF_8_BYTES, 10), entries); } public void testAcquireAndGet(ChronicleMap map, int entries) @@ -526,9 +532,9 @@ public class ChronicleMapTest { @Test public void testAcquireFromMultipleThreads() throws Exception { int entries = 1000 * 1000; - testAcquireFromMultipleThreads(getSharedMap(entries, 128, 24, NO_ALIGNMENT)); - testAcquireFromMultipleThreads(getSharedMap(entries, 128, 24, OF_4_BYTES)); - testAcquireFromMultipleThreads(getSharedMap(entries, 128, 24, OF_8_BYTES)); + testAcquireFromMultipleThreads(getSharedMap(entries, 128, NO_ALIGNMENT, 10)); + testAcquireFromMultipleThreads(getSharedMap(entries, 128, OF_4_BYTES, 10)); + testAcquireFromMultipleThreads(getSharedMap(entries, 128, OF_8_BYTES, 10)); } public void testAcquireFromMultipleThreads(ChronicleMap map) @@ -582,7 +588,7 @@ public class ChronicleMapTest { .entries(entries) .actualSegments(8 * 1024) .entryAndValueAlignment(OF_8_BYTES) - .entrySize(24); + .keySize(13); File tmpFile = File.createTempFile("testAcquirePerf", ".deleteme"); tmpFile.deleteOnExit(); @@ -603,8 +609,10 @@ public class ChronicleMapTest { StringBuilder sb = new StringBuilder(); long next = 50 * 1000 * 1000; // use a factor to give up to 10 digit numbers. - int factor = Math.max(1, (int) ((10 * 1000 * 1000 * 1000L - 1) / entries)); - for (long j = t % independence; j < entries + independence - 1; j += independence) { + int factor = Math.max(1, + (int) ((10 * 1000 * 1000 * 1000L - 1) / entries)); + for (long j = t % independence; j < entries + independence - 1; + j += independence) { sb.setLength(0); sb.append("us:"); sb.append(j * factor); @@ -625,7 +633,8 @@ public class ChronicleMapTest { future.get(); } long time = System.currentTimeMillis() - start; - System.out.printf("Throughput %.1f M ops/sec%n", threads * entries / independence / 1000.0 / time); + System.out.printf("Throughput %.1f M ops/sec%n", + threads * entries / independence / 1000.0 / time); } printStatus(); File file = map.file(); @@ -654,7 +663,7 @@ public class ChronicleMapTest { .entries(entries) .entryAndValueAlignment(OF_8_BYTES) .actualSegments(256) - .entrySize(24); + .keySize(13); File tmpFile = File.createTempFile("testAcquirePerf", ".deleteme"); tmpFile.deleteOnExit(); @@ -676,8 +685,10 @@ public class ChronicleMapTest { long next = 50 * 1000 * 1000; Random rand = new Random(); // use a factor to give up to 10 digit numbers. - int factor = Math.max(1, (int) ((10 * 1000 * 1000 * 1000L - 1) / entries)); - for (long j = t % independence; j < entries + independence - 1; j += independence) { + int factor = Math.max(1, + (int) ((10 * 1000 * 1000 * 1000L - 1) / entries)); + for (long j = t % independence; j < entries + independence - 1; + j += independence) { sb.setLength(0); sb.append("us:"); sb.append(j * factor); @@ -707,7 +718,8 @@ public class ChronicleMapTest { future.get(); } long time = System.currentTimeMillis() - start; - System.out.printf("Throughput %.1f M ops/sec%n", threads * entries / independence / 1000.0 / time); + System.out.printf("Throughput %.1f M ops/sec%n", + threads * entries / independence / 1000.0 / time); } printStatus(); File file = map.file(); @@ -853,14 +865,16 @@ public class ChronicleMapTest { } @Test - public void testPutAndRemove() throws IOException, ClassNotFoundException, IllegalAccessException, InstantiationException { + public void testPutAndRemove() throws IOException, ClassNotFoundException, + IllegalAccessException, InstantiationException { String TMP = System.getProperty("java.io.tmpdir"); int entries = 100 * 1000; ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) .entries(entries) .minSegments(16) - .entrySize(32) + .keySize("user:".length() + 6) + .valueSize("value:".length() + 6) .putReturnsNull(true) .removeReturnsNull(true).create(); StringBuilder key = new StringBuilder(); @@ -1412,7 +1426,7 @@ public class ChronicleMapTest { final ChronicleMapBuilder builder = ChronicleMapBuilder .of(CharSequence.class, LongValue.class) .entries(1000) - .entrySize(16); + .keySize("x".length()); final ChronicleMap map = builder.create(); @@ -1425,7 +1439,7 @@ public class ChronicleMapTest { ChronicleMapBuilder builder = ChronicleMapBuilder .of(CharSequence.class, LongValue.class) .entries(1000) - .entrySize(16); + .keySize("one".length()); final ChronicleMap map = builder.create(); @@ -1483,8 +1497,7 @@ public class ChronicleMapTest { ChronicleMapBuilder builder = ChronicleMapBuilder .of(CharSequence.class, String.class) .entries(1000) - .defaultValue("") - .entrySize(16); + .defaultValue(""); try (final ChronicleMap map = builder.create()) { // this will add the entry @@ -1500,7 +1513,8 @@ public class ChronicleMapTest { try (final ChronicleMap map = ChronicleMapBuilder .of(CharSequence.class, CharSequence.class) .entries(1000) - .entrySize(40) + .keySize("one".length()) + .valueSize("Hello World".length()) .defaultValue("") .create()) { StringBuilder value = new StringBuilder(); @@ -1518,7 +1532,7 @@ public class ChronicleMapTest { ChronicleMapBuilder builder = ChronicleMapBuilder .of(CharSequence.class, LongValue.class) .entries(1000) - .entrySize(40); + .keySize("one".length()); File tmpFile = File.createTempFile("testAcquireUsingLocked", ".deleteme"); tmpFile.deleteOnExit(); final ChronicleMap map = builder.createPersistedTo(tmpFile); diff --git a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java index 89290d4f8..fc53b204a 100644 --- a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java @@ -34,6 +34,8 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicLong; +import static java.lang.Math.log10; +import static java.lang.Math.round; import static net.openhft.chronicle.map.Alignment.OF_4_BYTES; import static net.openhft.lang.model.DataValueClasses.newDirectReference; import static org.junit.Assert.assertTrue; @@ -51,12 +53,13 @@ public class EntryCountMapTest { } private static ChronicleMap getSharedMap( - long entries, int segments, int entrySize) throws IOException { - ChronicleMapBuilder mapBuilder = ChronicleMapBuilder.of(CharSequence - .class, LongValue.class) + long entries, int segments, int keySize) throws IOException { + ChronicleMapBuilder mapBuilder = + ChronicleMapBuilder.of(CharSequence.class, LongValue.class) .entries(entries) .actualSegments(segments) - .entrySize(entrySize); + .keySize(keySize) + .maxEntryOversizeFactor(1); return mapBuilder.createPersistedTo(getPersistenceFile()); } @@ -182,12 +185,17 @@ public class EntryCountMapTest { System.out.printf("Score: %.2f%n", scoreCount / score); } - private Future testEntriesMaxSize(ExecutorService es, final int segments, final int minSize, final int maxSize, final int seed) throws IOException { + private Future testEntriesMaxSize(ExecutorService es, final int segments, + final int minSize, final int maxSize, final int seed) + throws IOException { assert minSize <= maxSize; Random random = new Random(seed); int counter = minSize + random.nextInt(9999 + maxSize); final int stride = 1 + random.nextInt(100); - final ChronicleMap map = getSharedMap(minSize, segments, 32); + int maxKeySize = "key:".length() + + (int) round(log10(moreThanMaxSize(maxSize) * stride + counter)) + 1; + final ChronicleMap map = + getSharedMap(minSize, segments, maxKeySize); return es.submit(new Callable() { @Override public Void call() throws Exception { @@ -202,14 +210,17 @@ public class EntryCountMapTest { Random random = new Random(seed); int counter = minSize + random.nextInt(9999 + maxSize); int stride = 1 + random.nextInt(100); - ChronicleMap map = getSharedMap(minSize, segments, 32); + int maxKeySize = "key:".length() + + (int) round(log10(moreThanMaxSize(maxSize) * stride + counter)) + 1; + ChronicleMap map = getSharedMap(minSize, segments, maxKeySize); testEntriesMaxSize0(segments, minSize, maxSize, counter, stride, map); } - void testEntriesMaxSize0(int segments, int minSize, int maxSize, int counter, int stride, ChronicleMap map) { + void testEntriesMaxSize0(int segments, int minSize, int maxSize, int counter, int stride, + ChronicleMap map) { LongValue longValue = newDirectReference(LongValue.class); try { - for (int j = 0; j < maxSize * 14 / 10 + 300; j++) { + for (int j = 0; j < moreThanMaxSize(maxSize); j++) { String key = "key:" + counter; counter += stride; // give a biased hashcode distribution. @@ -229,14 +240,20 @@ public class EntryCountMapTest { boolean condition = minSize <= map.size() && map.size() <= maxSize; if (!condition) { dumpMapStats(segments, minSize, map); - assertTrue("stride: " + stride + ", seg: " + segments + ", min: " + minSize + ", size: " + map.size(), condition); + assertTrue("stride: " + stride + ", seg: " + segments + ", min: " + minSize + + ", size: " + map.size(), condition); } } finally { map.close(); } } - private void dumpMapStats(int segments, int minSize, ChronicleMap map) { + private static int moreThanMaxSize(int maxSize) { + return maxSize * 14 / 10 + 300; + } + + private void dumpMapStats(int segments, int minSize, + ChronicleMap map) { long[] a = ((VanillaChronicleMap) map).segmentSizes(); System.out.println("segs: " + segments + " min: " + minSize + " was " + map.size() + " " + Arrays.toString(a) diff --git a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java index 8be61bd82..ab39ca043 100644 --- a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java +++ b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java @@ -50,7 +50,7 @@ public class MapByNameTest { final MapInstanceConfig config = (MapInstanceConfig) ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entrySize(900) + .keySize("hello".length()).valueSize("world".length()) .entries(2) .minSegments(2) .replication((byte) 1, TcpTransportAndNetworkConfig.of(8244)) diff --git a/src/test/java/net/openhft/chronicle/map/OSResizesMain.java b/src/test/java/net/openhft/chronicle/map/OSResizesMain.java index 7d3c986b5..758041c7f 100755 --- a/src/test/java/net/openhft/chronicle/map/OSResizesMain.java +++ b/src/test/java/net/openhft/chronicle/map/OSResizesMain.java @@ -33,13 +33,13 @@ public class OSResizesMain { public static void main(String[] args) throws IOException, InterruptedException { File file = File.createTempFile("over-sized", "deleteme"); int valueSize = 1000 * 1000; - ChronicleMap map = ChronicleMapBuilder.of(String.class, ByteArray.class) - .entrySize(50 * 1024 * 1024) - .valueSize(valueSize + 16) - .createPersistedTo(file); byte[] chars = new byte[valueSize]; - Arrays.fill(chars, (byte) '+'); ByteArray ba = new ByteArray(chars); + ChronicleMap map = ChronicleMapBuilder.of(String.class, ByteArray.class) + .keySize("key-".length() + 4) + .constantValueSizeBySample(ba) + .createPersistedTo(file); + Arrays.fill(chars, (byte) '+'); for (int i = 0; i < 1000; i++) { map.put("key-" + i, ba); } diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index e1b87eecc..81ad75cd2 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -465,22 +465,20 @@ public class StatelessClientTest { @Test(timeout = 10000) public void testLargeEntries() throws IOException, InterruptedException { - - - int keySize = 4; int valueSize = 1000000; - int entrySize = keySize + valueSize; char[] value = new char[valueSize]; Arrays.fill(value, 'X'); - try (ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entrySize(entrySize) + String sampleValue = new String(value); + try (ChronicleMap serverMap = + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .constantValueSizeBySample(sampleValue) .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) { try (ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer .class, CharSequence.class) - .entrySize(entrySize) + .constantValueSizeBySample(sampleValue) .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { // for (int i = 0; i < 128; i++) { diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationLargeEntryTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationLargeEntryTest.java index daffb7415..b000d7eba 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationLargeEntryTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationLargeEntryTest.java @@ -1,6 +1,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.lang.MemoryUnit; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -24,7 +25,7 @@ public class TcpReplicationLargeEntryTest { private ChronicleMap map2; static int s_port = 8010; - int valueSize = 1_00_000; // 1MB + int valueSize = (int) MemoryUnit.MEGABYTES.toBytes(1); char[] valueX = new char[valueSize - 100]; char[] valueY = new char[valueSize - 100]; @@ -34,9 +35,7 @@ public class TcpReplicationLargeEntryTest { Arrays.fill(valueX, 'X'); Arrays.fill(valueX, 'Y'); - int keySize = 4; - int entrySize = keySize + valueSize; - + String sampleValue = new String(valueX); final InetSocketAddress endpoint = new InetSocketAddress("localhost", s_port + 1); { @@ -47,7 +46,7 @@ public class TcpReplicationLargeEntryTest { map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entrySize(entrySize) + .constantValueSizeBySample(sampleValue) .entries(2) .actualSegments(1) .replication((byte) 1, tcpConfig1) @@ -62,7 +61,7 @@ public class TcpReplicationLargeEntryTest { .tcpBufferSize(1024 * 64); map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entrySize(entrySize) + .constantValueSizeBySample(sampleValue) .entries(2) .replication((byte) 2, tcpConfig2) .instance() diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationWithAcquireUsingLockedTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationWithAcquireUsingLockedTest.java index d802908fc..5a8bbeea7 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationWithAcquireUsingLockedTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationWithAcquireUsingLockedTest.java @@ -80,20 +80,22 @@ public class TcpReplicationWithAcquireUsingLockedTest { (CharSequence.class, TestInstrumentVOInterface .class) - .entries(5000L).entrySize(2000) + .entries(5000L).keySize("hello".length()) .replication((byte) 1, config1).create()) { map1a = map1; - TcpTransportAndNetworkConfig config2 = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)); + TcpTransportAndNetworkConfig config2 = TcpTransportAndNetworkConfig + .of(8077, new InetSocketAddress("127.0.0.1", 8076)); - try (ChronicleMap map2 = ChronicleMapBuilder.of(CharSequence.class, + try (ChronicleMap map2 = + ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class) .putReturnsNull(true) .removeReturnsNull(true) .replication((byte) 2, config2) - .entries(5000L).entrySize(2000).create()) { + .entries(5000L).keySize("hello".length()).create()) { map2a = map2; @@ -131,20 +133,22 @@ public class TcpReplicationWithAcquireUsingLockedTest { (CharSequence.class, TestInstrumentVOInterface .class) - .entries(5000L).entrySize(2000) + .entries(5000L).keySize("hello".length()) .replication((byte) 1, config1).create()) { map1a = map1; - TcpTransportAndNetworkConfig config2 = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)); + TcpTransportAndNetworkConfig config2 = + TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)); - try (ChronicleMap map2 = ChronicleMapBuilder.of(CharSequence.class, + try (ChronicleMap map2 = + ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class) .putReturnsNull(true) .removeReturnsNull(true) .replication((byte) 2, config2) - .entries(5000L).entrySize(2000).create()) { + .entries(5000L).keySize("hello".length()).create()) { map2a = map2; diff --git a/src/test/java/net/openhft/chronicle/map/example/ServersMapMain.java b/src/test/java/net/openhft/chronicle/map/example/ServersMapMain.java index 57706e3c0..b4fdd57b7 100644 --- a/src/test/java/net/openhft/chronicle/map/example/ServersMapMain.java +++ b/src/test/java/net/openhft/chronicle/map/example/ServersMapMain.java @@ -39,9 +39,11 @@ public class ServersMapMain { public static void startServer() throws IOException { File file = File.createTempFile("testServersMapMain", ".deleteme"); file.deleteOnExit(); - final ChronicleMap serverMap = ChronicleMapBuilder.of(byte[].class, byte[].class) + final ChronicleMap serverMap = + ChronicleMapBuilder.of(byte[].class, byte[].class) .putReturnsNull(true) - .entrySize(50) + .constantKeySizeBySample(new byte[8]) + .constantValueSizeBySample(new byte[32]) .replication((byte) 2, TcpTransportAndNetworkConfig.of(port)) .createPersistedTo(file); System.out.println("Server started"); @@ -59,7 +61,8 @@ public class ServersMapMain { } else { File file = File.createTempFile("testServersMapMain", ".deleteme"); file.deleteOnExit(); - TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port)); + TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port)); map = ChronicleMapBuilder.of( byte[].class, byte[].class) @@ -83,11 +86,13 @@ public class ServersMapMain { puts += t2 - t1; gets += t3 - t2; if (t2 - t1 > lastAveragePut * 100 || t3 - t2 > lastAverageGet * 100) - System.out.printf("Took put/get took %.1f/%.1f us%n", (t2 - t1) / 1e3, (t3 - t2) / 1e3); + System.out.printf("Took put/get took %.1f/%.1f us%n", + (t2 - t1) / 1e3, (t3 - t2) / 1e3); } lastAveragePut = puts / entries; lastAverageGet = gets / entries; - System.out.printf("Average took put/get took %.1f/%.1f us%n", lastAveragePut / 1e3, lastAverageGet / 1e3); + System.out.printf("Average took put/get took %.1f/%.1f us%n", + lastAveragePut / 1e3, lastAverageGet / 1e3); } } diff --git a/src/test/java/net/openhft/chronicle/map/example/StringsMapMain.java b/src/test/java/net/openhft/chronicle/map/example/StringsMapMain.java index 08b700c13..4d55a5bfc 100644 --- a/src/test/java/net/openhft/chronicle/map/example/StringsMapMain.java +++ b/src/test/java/net/openhft/chronicle/map/example/StringsMapMain.java @@ -38,10 +38,12 @@ public class StringsMapMain { public static void startServer() throws IOException { File file = File.createTempFile("testServersMapMain", ".deleteme"); file.deleteOnExit(); + int keyAndValueSize = (int) Math.round(Math.log10(entries)); final ChronicleMap serverMap = ChronicleMapBuilder .of(CharSequence.class, CharSequence.class) .putReturnsNull(true) - .entrySize(50) + .keySize(keyAndValueSize) + .valueSize(keyAndValueSize) .replication((byte) 2, TcpTransportAndNetworkConfig.of(port)) .createPersistedTo(file); System.out.println("Server started"); @@ -58,7 +60,8 @@ public class StringsMapMain { } else { File file = File.createTempFile("testServersMapMain", ".deleteme"); file.deleteOnExit(); - TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port)); + TcpTransportAndNetworkConfig tcpConfig = + TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port)); map = ChronicleMapBuilder .of(CharSequence.class, CharSequence.class) @@ -90,11 +93,13 @@ public class StringsMapMain { puts += t2 - t1; gets += t3 - t2; // if (t2 - t1 > lastAveragePut * 100 || t3 - t2 > lastAverageGet * 100) -// System.out.printf("Took put/get took %.1f/%.1f us%n", (t2 - t1) / 1e3, (t3 - t2) / 1e3); +// System.out.printf("Took put/get took %.1f/%.1f us%n", +// (t2 - t1) / 1e3, (t3 - t2) / 1e3); } lastAveragePut = puts / entries; lastAverageGet = gets / entries; - System.out.printf("Average took put/get took %.1f/%.1f us%n", lastAveragePut / 1e3, lastAverageGet / 1e3); + System.out.printf("Average took put/get took %.1f/%.1f us%n", + lastAveragePut / 1e3, lastAverageGet / 1e3); } } diff --git a/src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongCASLeft.java b/src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongCASLeft.java index cd1a98c5e..1b02f5a6a 100644 --- a/src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongCASLeft.java +++ b/src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongCASLeft.java @@ -50,7 +50,8 @@ public class PingPongCASLeft { playPingPong(chm, 4, 5, true, "PingPongCASLEFT"); } - static void playPingPong(ChronicleMap chm, double _coupon, double _coupon2, boolean setFirst, final String desc) { + static void playPingPong(ChronicleMap chm, double _coupon, + double _coupon2, boolean setFirst, final String desc) { BondVOInterface bond1 = newDirectReference(BondVOInterface.class); BondVOInterface bond2 = newDirectReference(BondVOInterface.class); BondVOInterface bond3 = newDirectReference(BondVOInterface.class); @@ -91,6 +92,6 @@ public class PingPongCASLeft { AffinitySupport.setThreadId(); return ChronicleMapBuilder.of(String.class, BondVOInterface.class) .entries(16) - .entrySize(64).create(); + .keySize("369604101".length()).create(); } } \ No newline at end of file diff --git a/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java b/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java index cecae189c..3a861300d 100644 --- a/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java +++ b/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java @@ -137,7 +137,7 @@ public class ProcessInstanceLimiter implements Runnable { ChronicleMapBuilder builder = ChronicleMapBuilder.of(String.class, Data.class); builder.entries(1000); - builder.entrySize(1024); + builder.keySize((DEFAULT_PROCESS_NAME + "#").length()); this.theSharedMap = builder.createPersistedTo(new File(sharedMapPath)); Thread t = new Thread(this, "ProcessInstanceLimiter updater"); t.setDaemon(true); From 4a05e33d46b5ef50e35f0107bdc8c9530e62b54d Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 19 Dec 2014 21:32:45 +0000 Subject: [PATCH 2/7] Add an example for lots of entries. --- .../chronicle/map/ChronicleMapTest.java | 3 +- .../map/example/LotsOfEntriesMain.java | 61 +++++++++++++++++++ 2 files changed, 62 insertions(+), 2 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index ea9a9ba7d..f9cf71bff 100755 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -523,7 +523,6 @@ public class ChronicleMapTest { map.close(); } - @Ignore("JIRA raised HCOLL-221") @Test public void testAcquireFromMultipleThreads() throws Exception { int entries = 1000 * 1000; @@ -538,7 +537,7 @@ public class ChronicleMapTest { CharSequence key = getUserCharSequence(0); map.acquireUsing(key, new LongValue$$Native()); - int iterations = 1000; + int iterations = 10000; int noOfThreads = 10; CyclicBarrier barrier = new CyclicBarrier(noOfThreads); diff --git a/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java b/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java new file mode 100644 index 000000000..67ffcec9c --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java @@ -0,0 +1,61 @@ +package net.openhft.chronicle.map.example; + +import net.openhft.chronicle.map.ChronicleMap; +import net.openhft.chronicle.map.ChronicleMapBuilder; +import net.openhft.lang.model.constraints.MaxSize; + +import java.io.File; +import java.io.IOException; +import java.util.Random; + +/** + * Created by peter on 19/12/14. + */ +public class LotsOfEntriesMain { + public static void main(String[] args) throws IOException { + workEntries(true); + workEntries(false); + } + + private static void workEntries(boolean add) throws IOException { + long entries = 100_000_000; + File file = new File("/tmp/lotsOfEntries.dat"); + ChronicleMap map = ChronicleMapBuilder + .of(CharSequence.class, MyFloats.class) + .entries(entries) + .entrySize(80) + .createPersistedTo(file); + Random rand = new Random(); + StringBuilder sb = new StringBuilder(); + MyFloats mf = map.newValueInstance(); + if (add) + for (int i = 0; i < 6; i++) + mf.setValueAt(i, i); + long start = System.nanoTime(); + for (long i = 0; i < entries; i++) { + sb.setLength(0); + int length = (int) (24 / (rand.nextFloat() + 24.0 / 1000)); + if (length > 2000) + throw new AssertionError(); + sb.append(i); + while (sb.length() < length) + sb.append("-key"); + if (add) + map.put(sb, mf); + else + map.getUsing(sb, mf); + if (i << -7 == 0 && i % 2000000 == 0) + System.out.println(i); + } + long time = System.nanoTime() - start; + System.out.printf("Map.size: %,d took an average of %,d ns to %s.%n", + map.size(), time / entries, add ? "add" : "get"); + map.close(); + } +} + +interface MyFloats { + public void setValueAt(@MaxSize(6) int index, float f); + + public float getValueAt(int index); +} From 66b33e3896d4ecd1964b503ce154f61b7e1e5780 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 19 Dec 2014 22:11:19 +0000 Subject: [PATCH 3/7] Test for lots of variable sized entries --- .../map/example/LotsOfEntriesMain.java | 83 ++++++++++++------- 1 file changed, 54 insertions(+), 29 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java b/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java index 67ffcec9c..feb808c9f 100644 --- a/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java +++ b/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java @@ -6,50 +6,75 @@ import net.openhft.lang.model.constraints.MaxSize; import java.io.File; import java.io.IOException; +import java.util.ArrayList; +import java.util.List; import java.util.Random; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; /** * Created by peter on 19/12/14. */ public class LotsOfEntriesMain { - public static void main(String[] args) throws IOException { + public static void main(String[] args) throws IOException, ExecutionException, InterruptedException { workEntries(true); workEntries(false); } - private static void workEntries(boolean add) throws IOException { - long entries = 100_000_000; + private static void workEntries(final boolean add) throws IOException, ExecutionException, InterruptedException { + final long entries = 100_000_000; File file = new File("/tmp/lotsOfEntries.dat"); - ChronicleMap map = ChronicleMapBuilder + final ChronicleMap map = ChronicleMapBuilder .of(CharSequence.class, MyFloats.class) - .entries(entries) - .entrySize(80) + .entries(entries*6/5) + .entrySize(256) .createPersistedTo(file); - Random rand = new Random(); - StringBuilder sb = new StringBuilder(); - MyFloats mf = map.newValueInstance(); - if (add) - for (int i = 0; i < 6; i++) - mf.setValueAt(i, i); - long start = System.nanoTime(); - for (long i = 0; i < entries; i++) { - sb.setLength(0); - int length = (int) (24 / (rand.nextFloat() + 24.0 / 1000)); - if (length > 2000) - throw new AssertionError(); - sb.append(i); - while (sb.length() < length) - sb.append("-key"); - if (add) - map.put(sb, mf); - else - map.getUsing(sb, mf); - if (i << -7 == 0 && i % 2000000 == 0) - System.out.println(i); + int threads = Runtime.getRuntime().availableProcessors(); + ExecutorService es = Executors.newFixedThreadPool(threads); + long block = (entries + threads - 1) / threads; + + final long start = System.nanoTime(); + List> futures = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + final long startI = t * block; + final long endI = Math.min((t + 1) * block, entries); + futures.add(es.submit(new Runnable() { + @Override + public void run() { + Random rand = new Random(startI); + StringBuilder sb = new StringBuilder(); + MyFloats mf = map.newValueInstance(); + if (add) + for (int i = 0; i < 6; i++) + mf.setValueAt(i, i); + for (long i = startI; i < endI; i++) { + sb.setLength(0); + int length = (int) (24 / (rand.nextFloat() + 24.0 / 1000)); + sb.append(i); + while (sb.length() < length) + sb.append("-key"); + try { + if (add) + map.put(sb, mf); + else + map.getUsing(sb, mf); + } catch (Exception e) { + System.out.println("map.size: "+map.size()); + throw e; + } + } + } + })); + } + for (Future future : futures) { + future.get(); } long time = System.nanoTime() - start; - System.out.printf("Map.size: %,d took an average of %,d ns to %s.%n", - map.size(), time / entries, add ? "add" : "get"); + es.shutdown(); + System.out.printf("Map.size: %,d with a throughput of %.1f million/sec to %s.%n", + map.size(), entries * 1e3 / time, add ? "add" : "get"); map.close(); } } From e3addd2951bb6460358896888ed3d67fe7080818 Mon Sep 17 00:00:00 2001 From: leventov Date: Sat, 20 Dec 2014 17:56:41 +0700 Subject: [PATCH 4/7] Tuned LotsOfEntriesMain --- .../map/example/LotsOfEntriesMain.java | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java b/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java index feb808c9f..c69210304 100644 --- a/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java +++ b/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java @@ -18,18 +18,28 @@ import java.util.concurrent.Future; * Created by peter on 19/12/14. */ public class LotsOfEntriesMain { - public static void main(String[] args) throws IOException, ExecutionException, InterruptedException { + public static void main(String[] args) + throws IOException, ExecutionException, InterruptedException { workEntries(true); workEntries(false); } - private static void workEntries(final boolean add) throws IOException, ExecutionException, InterruptedException { + private static void workEntries(final boolean add) + throws IOException, ExecutionException, InterruptedException { final long entries = 100_000_000; File file = new File("/tmp/lotsOfEntries.dat"); + double averageEntrySize = + 1 + // key size marshalling size + // + 2 is average oversize because we append 4-letter "-key" in a loop + (Math.log(1.024) - Math.log(0.024)) * 24 + 2 + // average key size + 24 + // value size + 2 // average loss due to value alignment. defaults to OF_4_BYTES because + // there are floats in values + ; final ChronicleMap map = ChronicleMapBuilder .of(CharSequence.class, MyFloats.class) - .entries(entries*6/5) - .entrySize(256) + .entries((long) (entries * (averageEntrySize / 16.0))) + .entrySize(16) // "chunk mode" .createPersistedTo(file); int threads = Runtime.getRuntime().availableProcessors(); ExecutorService es = Executors.newFixedThreadPool(threads); From f9b4ba79e56495914ed6558698c2ee7f8c8755e0 Mon Sep 17 00:00:00 2001 From: leventov Date: Sat, 20 Dec 2014 18:10:12 +0700 Subject: [PATCH 5/7] Account internal fragmentation in LotsOfEntriesMain --- .../openhft/chronicle/map/example/LotsOfEntriesMain.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java b/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java index c69210304..dc2f4a151 100644 --- a/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java +++ b/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java @@ -38,8 +38,11 @@ public class LotsOfEntriesMain { ; final ChronicleMap map = ChronicleMapBuilder .of(CharSequence.class, MyFloats.class) - .entries((long) (entries * (averageEntrySize / 16.0))) - .entrySize(16) // "chunk mode" + .entries((long) (entries * + (averageEntrySize / 8.0 + + 0.5))) // account average internal fragmentation -- + // because we need to round up to integral number of chunks + .entrySize(8) // "chunk mode" .createPersistedTo(file); int threads = Runtime.getRuntime().availableProcessors(); ExecutorService es = Executors.newFixedThreadPool(threads); From 8d44587c7c3a54fc173d455ef6769cca98a00d4a Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Sat, 20 Dec 2014 13:27:59 +0000 Subject: [PATCH 6/7] HCOLL-197 - improved buffer resizing --- .../hash/replication/UdpTransportConfig.java | 38 +++++-- .../map/AbstractChannelReplicator.java | 6 +- .../chronicle/map/ChannelProvider.java | 5 +- .../openhft/chronicle/map/Replicators.java | 98 ++++++++++++++++--- .../chronicle/map/StatelessChronicleMap.java | 52 +++++++--- .../openhft/chronicle/map/TcpReplicator.java | 95 +++++++++++------- .../openhft/chronicle/map/UdpReplicator.java | 21 ++-- .../openhft/chronicle/map/WriterWithSize.java | 37 ++++++- .../openhft/chronicle/map/SerializerTest.java | 6 +- 9 files changed, 267 insertions(+), 91 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java b/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java index b32325673..b253e7bff 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java @@ -25,16 +25,26 @@ import java.net.InetAddress; import java.net.NetworkInterface; public final class UdpTransportConfig { + public static final int DEFAULT_UDP_BUFFER_SIZE = 64 * 1024; private final InetAddress address; private final int port; - private final @Nullable NetworkInterface networkInterface; + + @Nullable + private final + NetworkInterface networkInterface; + + @NotNull private final ThrottlingConfig throttlingConfig; + private final int udpBufferSize; + private UdpTransportConfig( InetAddress address, int port, @Nullable NetworkInterface networkInterface, - ThrottlingConfig throttlingConfig) { + ThrottlingConfig throttlingConfig, + int udpBufferSize) { + this.udpBufferSize = udpBufferSize; if (address == null) { throw new NullPointerException("Null address"); } @@ -49,22 +59,23 @@ public final class UdpTransportConfig { } public static UdpTransportConfig multiCast(@NotNull InetAddress address, int port, - @NotNull NetworkInterface networkInterface) { + @Nullable NetworkInterface networkInterface) { if (!address.isMulticastAddress() || networkInterface == null) throw new IllegalArgumentException(); - return create(address, port, networkInterface, ThrottlingConfig.noThrottling()); + return create(address, port, networkInterface, ThrottlingConfig.noThrottling(), DEFAULT_UDP_BUFFER_SIZE); } public static UdpTransportConfig of(@NotNull InetAddress address, int port) { if (address.isMulticastAddress()) throw new IllegalArgumentException(); - return create(address, port, null, ThrottlingConfig.noThrottling()); + return create(address, port, null, ThrottlingConfig.noThrottling(), DEFAULT_UDP_BUFFER_SIZE); } static UdpTransportConfig create(InetAddress address, int port, @Nullable NetworkInterface networkInterface, - ThrottlingConfig throttlingConfig) { - return new UdpTransportConfig(address, port, networkInterface, throttlingConfig); + ThrottlingConfig throttlingConfig, + final int udpBufferSize1) { + return new UdpTransportConfig(address, port, networkInterface, throttlingConfig, udpBufferSize1); } @NotNull @@ -129,11 +140,20 @@ public final class UdpTransportConfig { @Nullable NetworkInterface networkInterface) { if (!address().isMulticastAddress()) throw new IllegalArgumentException(); - return create(address(), port(), networkInterface, throttlingConfig()); + return create(address(), port(), networkInterface, throttlingConfig(), udpBufferSize()); } public UdpTransportConfig throttlingConfig(@NotNull ThrottlingConfig throttlingConfig) { ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "UDP"); - return create(address(), port(), networkInterface(), throttlingConfig); + return create(address(), port(), networkInterface(), throttlingConfig, udpBufferSize()); } + + public int udpBufferSize() { + return udpBufferSize; + } + + public UdpTransportConfig udpBufferSize(int udpBufferSize) { + return create(address(), port(), networkInterface(), throttlingConfig(), udpBufferSize); + } + } diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 94c8ab798..4834549f2 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -410,7 +410,7 @@ abstract class AbstractChannelReplicator implements Closeable { } } - static class EntryCallback extends Replica.EntryCallback { + static class EntryCallback extends Replica.EntryCallback implements BufferResizer { private final Replica.EntryExternalizable externalizable; @@ -438,7 +438,7 @@ abstract class AbstractChannelReplicator implements Closeable { } - private void resizeBuffer(int size) { + public Bytes resizeBuffer(int size) { if (LOG.isDebugEnabled()) LOG.debug("resizing buffer to size=" + size); @@ -474,7 +474,7 @@ abstract class AbstractChannelReplicator implements Closeable { assert out.capacity() == size; assert out.capacity() == in.capacity(); assert in.limit() == in.capacity(); - + return in; } public boolean shouldBeIgnored(final Bytes entry, final int chronicleId) { diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index b69f8c3bb..e24bf5890 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -60,11 +60,14 @@ final class ChannelProvider implements Closeable { return channelProvider; channelProvider = new ChannelProvider(hub); TcpTransportAndNetworkConfig tcpConfig = hub.tcpTransportAndNetwork(); + + if (tcpConfig != null) { + final TcpReplicator tcpReplicator = new TcpReplicator( channelProvider.asReplica, channelProvider.asEntryExternalizable, - tcpConfig, null, hub.remoteNodeValidator()); + tcpConfig, hub.remoteNodeValidator(), null); channelProvider.add(tcpReplicator); } diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index f1c9aa232..e3f510797 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -19,43 +19,113 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.AbstractReplication; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.hash.replication.UdpTransportConfig; +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.Bytes; import org.jetbrains.annotations.NotNull; import java.io.Closeable; import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; final class Replicators { static final String ONLY_UDP_WARN_MESSAGE = "MISSING TCP REPLICATION : The UdpReplicator only attempts to read data " + - "(it does not enforce or guarantee delivery), you should use" + - "the UdpReplicator if you have a large number of nodes, and you wish" + - "to receive the data before it becomes available on TCP/IP. Since data" + - "delivery is not guaranteed, it is recommended that you only use" + - "the UDP Replicator in conjunction with a TCP Replicator"; + "(it does not enforce or guarantee delivery), you should use" + + "the UdpReplicator if you have a large number of nodes, and you wish" + + "to receive the data before it becomes available on TCP/IP. Since data" + + "delivery is not guaranteed, it is recommended that you only use" + + "the UDP Replicator in conjunction with a TCP Replicator"; private Replicators() { } + + static class OutBuffer implements BufferResizer { + + @NotNull + private ByteBufferBytes in; + + @NotNull + private ByteBuffer out; + + @NotNull + public ByteBufferBytes in() { + return in; + } + + @NotNull + public ByteBuffer out() { + return out; + } + + + OutBuffer(final int tcpBufferSize) { + out = ByteBuffer.allocateDirect(tcpBufferSize); + in = new ByteBufferBytes(out); + } + + @Override + public Bytes resizeBuffer(int newCapacity) { + + + if (newCapacity < out.capacity()) + throw new IllegalStateException("it not possible to resize the buffer smaller"); + + assert newCapacity < Integer.MAX_VALUE; + + final ByteBuffer result = ByteBuffer.allocate(newCapacity).order(ByteOrder.nativeOrder()); + final long bytesPosition = in.position(); + + in = new ByteBufferBytes(result.slice()); + + out.position(0); + out.limit((int) bytesPosition); + + int numberOfLongs = (int) bytesPosition / 8; + + // chunk in longs first + for (int i = 0; i < numberOfLongs; i++) { + in.writeLong(out.getLong()); + } + + for (int i = numberOfLongs * 8; i < bytesPosition; i++) { + in.writeByte(out.get()); + } + + out = result; + + assert out.capacity() == in.capacity(); + + assert out.capacity() == newCapacity; + assert out.capacity() == in.capacity(); + assert in.limit() == in.capacity(); + return in; + + } + } + static Replicator tcp(final AbstractReplication replication) { return new Replicator() { + + @Override protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder, @NotNull final Replica replica, @NotNull final Replica.EntryExternalizable entryExternalizable, - final ChronicleMap chronicleMap) - throws IOException { - - StatelessServerConnector statelessServer = new StatelessServerConnector( - builder.keyBuilder, builder.valueBuilder, - (VanillaChronicleMap) chronicleMap, builder.entrySize(true)); + final ChronicleMap chronicleMap) throws IOException { TcpTransportAndNetworkConfig tcpConfig = replication.tcpTransportAndNetwork(); - assert tcpConfig != null; + + TcpReplicator.StatelessClientParameters statelessClientParameters = new TcpReplicator.StatelessClientParameters( + (VanillaChronicleMap) chronicleMap, + builder.keyBuilder, + builder.valueBuilder); + return new TcpReplicator(replica, entryExternalizable, tcpConfig, - statelessServer, - replication.remoteNodeValidator()); + replication.remoteNodeValidator(), statelessClientParameters); } }; } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index c3c012792..f8f4cf2c5 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -139,10 +139,22 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon this.config = config; - keyReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.keyBuilder); - keyWriterWithSize = new WriterWithSize<>(chronicleMapBuilder.keyBuilder); - valueReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.valueBuilder); - valueWriterWithSize = new WriterWithSize<>(chronicleMapBuilder.valueBuilder); + final BufferResizer outBufferResizer = new BufferResizer() { + @Override + public Bytes resizeBuffer(int newCapacity) { + return resizeBufferOutBuffer(newCapacity); + + } + }; + + keyReaderWithSize = new ReaderWithSize(chronicleMapBuilder.keyBuilder); + keyWriterWithSize = new WriterWithSize(chronicleMapBuilder.keyBuilder, outBufferResizer); + + valueReaderWithSize = new ReaderWithSize(chronicleMapBuilder.valueBuilder); + + valueWriterWithSize = new WriterWithSize(chronicleMapBuilder.valueBuilder, + outBufferResizer); + this.putReturnsNull = chronicleMapBuilder.putReturnsNull(); this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull(); this.maxEntrySize = chronicleMapBuilder.entrySize(true); @@ -515,21 +527,25 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return (long) ((double) outBytes.position() / percentageComplete); } - private void resizeBufferOutBuffer(int size, long start) { + private Bytes resizeBufferOutBuffer(int newCapacity) { + return resizeBufferOutBuffer(newCapacity, outBytes.position()); + } + + private Bytes resizeBufferOutBuffer(int newCapacity, long start) { assert outBytesLock.isHeldByCurrentThread(); assert !inBytesLock.isHeldByCurrentThread(); if (LOG.isDebugEnabled()) - LOG.debug("resizing buffer to size=" + size + " ,name=" + name); + LOG.debug("resizing buffer to newCapacity=" + newCapacity + " ,name=" + name); - if (size < outBuffer.capacity()) + if (newCapacity < outBuffer.capacity()) throw new IllegalStateException("it not possible to resize the buffer smaller"); - assert size < Integer.MAX_VALUE; + assert newCapacity < Integer.MAX_VALUE; - final ByteBuffer result = ByteBuffer.allocate(size).order(ByteOrder.nativeOrder()); + final ByteBuffer result = ByteBuffer.allocate(newCapacity).order(ByteOrder.nativeOrder()); final long bytesPosition = outBytes.position(); outBytes = new ByteBufferBytes(result.slice()); @@ -552,25 +568,27 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon assert outBuffer.capacity() == outBytes.capacity(); - assert outBuffer.capacity() == size; + assert outBuffer.capacity() == newCapacity; assert outBuffer.capacity() == outBytes.capacity(); assert outBytes.limit() == outBytes.capacity(); outBytes.position(start); + return outBytes; } - private void resizeBufferInBuffer(int size, long start) { + + private void resizeBufferInBuffer(int newCapacity, long start) { assert !outBytesLock.isHeldByCurrentThread(); assert inBytesLock.isHeldByCurrentThread(); if (LOG.isDebugEnabled()) - LOG.debug("InBuffer resizing buffer to size=" + size + " ,name=" + name); + LOG.debug("InBuffer resizing buffer to newCapacity=" + newCapacity + " ,name=" + name); - if (size < inBuffer.capacity()) + if (newCapacity < inBuffer.capacity()) throw new IllegalStateException("it not possible to resize the buffer smaller"); - assert size < Integer.MAX_VALUE; + assert newCapacity < Integer.MAX_VALUE; - final ByteBuffer result = ByteBuffer.allocate(size).order(ByteOrder.nativeOrder()); + final ByteBuffer result = ByteBuffer.allocate(newCapacity).order(ByteOrder.nativeOrder()); final long bytesPosition = inBytes.position(); inBytes = new ByteBufferBytes(result.slice()); @@ -592,7 +610,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon inBuffer = result; assert inBuffer.capacity() == inBytes.capacity(); - assert inBuffer.capacity() == size; + assert inBuffer.capacity() == newCapacity; assert inBuffer.capacity() == inBytes.capacity(); assert inBytes.limit() == inBytes.capacity(); @@ -1370,6 +1388,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private void resizeToMessageOutBuffer(long start, @NotNull IllegalArgumentException e) { + + e.printStackTrace(); assert outBytesLock.isHeldByCurrentThread(); assert !inBytesLock.isHeldByCurrentThread(); String message = e.getMessage(); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 66bb147d4..b0c3a1786 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -62,7 +62,7 @@ interface Work { * * @author Rob Austin. */ -final class TcpReplicator extends AbstractChannelReplicator implements Closeable { +final class TcpReplicator extends AbstractChannelReplicator implements Closeable { public static final long TIMESTAMP_FACTOR = 10000; private static final int STATELESS_CLIENT = -127; @@ -86,8 +86,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private final Replica.EntryExternalizable externalizable; @NotNull private final TcpTransportAndNetworkConfig replicationConfig; - @Nullable - private final StatelessServerConnector statelessServerConnector; + + private final @Nullable RemoteNodeValidator remoteNodeValidator; @@ -95,19 +95,34 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private long selectorTimeout; + StatelessClientParameters statelessClientParameters; + + static class StatelessClientParameters { + public StatelessClientParameters(VanillaChronicleMap map, + SerializationBuilder keySerializationBuilder, + SerializationBuilder valueSerializationBuilder) { + this.map = map; + this.keySerializationBuilder = keySerializationBuilder; + this.valueSerializationBuilder = valueSerializationBuilder; + } + + VanillaChronicleMap map; + SerializationBuilder keySerializationBuilder; + SerializationBuilder valueSerializationBuilder; + } + /** - * @param statelessServerConnector set to NULL if not required * @throws IOException on an io error. */ public TcpReplicator(@NotNull final Replica replica, @NotNull final Replica.EntryExternalizable externalizable, @NotNull final TcpTransportAndNetworkConfig replicationConfig, - @Nullable StatelessServerConnector statelessServerConnector, - @Nullable RemoteNodeValidator remoteNodeValidator) throws IOException { + @Nullable final RemoteNodeValidator remoteNodeValidator, + @Nullable final StatelessClientParameters statelessClientParameters) throws IOException { - super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig() - ); - this.statelessServerConnector = statelessServerConnector; + super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig()); + + this.statelessClientParameters = statelessClientParameters; final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig(); long throttleBucketInterval = throttlingConfig.bucketInterval(MILLISECONDS); @@ -881,7 +896,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable /** * Attached to the NIO selection key via methods such as {@link SelectionKey#attach(Object)} */ - class Attached implements Replica.ModificationNotifier { + class Attached implements Replica.ModificationNotifier { public TcpSocketChannelEntryReader entryReader; public TcpSocketChannelEntryWriter entryWriter; @@ -901,16 +916,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable return handShakingComplete; } - void clearHandShaking() { - handShakingComplete = false; - - remoteIdentifier = Byte.MIN_VALUE; - remoteBootstrapTimestamp = Long.MIN_VALUE; - remoteHeartbeatInterval = heartBeatIntervalMillis; - hasRemoteHeartbeatInterval = false; - remoteModificationIterator = null; - } - /** * called whenever there is a change to the modification iterator */ @@ -934,9 +939,17 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable @Nullable public Work uncompletedWork; private long lastSentTime; + StatelessServerConnector statelessServer; private TcpSocketChannelEntryWriter() { entryCallback = new EntryCallback(externalizable, replicationConfig.tcpBufferSize()); + + if (statelessClientParameters != null) + statelessServer = new StatelessServerConnector( + statelessClientParameters.map, entryCallback, + replicationConfig.tcpBufferSize(), + statelessClientParameters.keySerializationBuilder, + statelessClientParameters.valueSerializationBuilder); } public boolean isWorkIncomplete() { @@ -960,6 +973,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable return entryCallback.in(); } + private ByteBuffer out() { + return entryCallback.out(); + } + /** * sends the identity and timestamp of this node to a remote node * @@ -1071,9 +1088,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable return len; } - private ByteBuffer out() { - return entryCallback.out(); - } /** * used to send an single zero byte if we have not send any data for up to the @@ -1128,7 +1142,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable /** * Reads map entries from a socket, this could be a client or server socket */ - class TcpSocketChannelEntryReader { + class TcpSocketChannelEntryReader { public static final int HEADROOM = 1024; public long lastHeartBeatReceived = System.currentTimeMillis(); ByteBuffer in; @@ -1238,6 +1252,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable boolean isStatelessClient = (state != 1); if (isStatelessClient) { + + final StatelessServerConnector statelessServerConnector = attached + .entryWriter.statelessServer; + if (statelessServerConnector == null) { LOG.error("", new IllegalArgumentException("received an event " + "from a stateless map, stateless maps are not " + @@ -1338,26 +1356,33 @@ class StatelessServerConnector { @NotNull private final ReaderWithSize keyReaderWithSize; + @NotNull private final WriterWithSize keyWriterWithSize; + @NotNull private final ReaderWithSize valueReaderWithSize; + @NotNull private final WriterWithSize valueWriterWithSize; + @NotNull private final VanillaChronicleMap map; - private final double maxEntrySizeBytes; + private final int tcpBufferSize; + StatelessServerConnector( - SerializationBuilder keySerializationBuilder, - SerializationBuilder valueSerializationBuilder, - @NotNull VanillaChronicleMap map, int maxEntrySizeBytes) { + @NotNull VanillaChronicleMap map, + @NotNull final BufferResizer bufferResizer, int tcpBufferSize, + final SerializationBuilder keySerializationBuilder, + final SerializationBuilder valueSerializationBuilder) { + this.tcpBufferSize = tcpBufferSize; + keyReaderWithSize = new ReaderWithSize<>(keySerializationBuilder); - keyWriterWithSize = new WriterWithSize<>(keySerializationBuilder); + keyWriterWithSize = new WriterWithSize<>(keySerializationBuilder, bufferResizer); valueReaderWithSize = new ReaderWithSize<>(valueSerializationBuilder); - valueWriterWithSize = new WriterWithSize<>(valueSerializationBuilder); + valueWriterWithSize = new WriterWithSize<>(valueSerializationBuilder, bufferResizer); this.map = map; - this.maxEntrySizeBytes = maxEntrySizeBytes; } @Nullable @@ -1784,7 +1809,7 @@ class StatelessServerConnector { while (iterator.hasNext()) { // we've filled up the buffer, so lets give another channel a chance to send // some data, we don't know the max key size, we will use the entrySize instead - if (out.remaining() <= maxEntrySizeBytes) { + if (out.position() > tcpBufferSize) { writeHeader(out, sizeLocation, count, true); return false; } @@ -1826,7 +1851,7 @@ class StatelessServerConnector { while (iterator.hasNext()) { // we've filled up the buffer, so lets give another channel a chance to send // some data, we don't know the max key size, we will use the entrySize instead - if (out.remaining() <= maxEntrySizeBytes) { + if (out.position() > tcpBufferSize) { writeHeader(out, sizeLocation, count, true); return false; } @@ -1859,7 +1884,7 @@ class StatelessServerConnector { return new Work() { @Override public boolean doWork(@NotNull Bytes out) { - if (out.remaining() <= maxEntrySizeBytes) + if (out.position() > tcpBufferSize) return false; final long sizeLocation = header(out, transactionId); @@ -1873,7 +1898,7 @@ class StatelessServerConnector { while (iterator.hasNext()) { // we've filled up the buffer, so lets give another channel a chance to send // some data, we don't know the max key size, we will use the entrySize instead - if (out.remaining() <= maxEntrySizeBytes) { + if (out.position() > tcpBufferSize) { writeHeader(out, sizeLocation, count, true); return false; } diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index a4cb64688..6060abb14 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java @@ -39,6 +39,9 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi private static final Logger LOG = LoggerFactory.getLogger(UdpReplicator.class.getName()); + public static final int UPD_BUFFER_SIZE = 64 * 1204; + @NotNull + private final UdpTransportConfig replicationConfig; public UdpReplicator(@NotNull final Replica replica, @NotNull final Replica.EntryExternalizable entryExternalizable, @@ -47,14 +50,16 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi throws IOException { super(replicationConfig, serializedEntrySize, replica.identifier()); + this.replicationConfig = replicationConfig; Replica.ModificationIterator modificationIterator = replica.acquireModificationIterator( ChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID, this); setReader(new UdpSocketChannelEntryReader(serializedEntrySize, entryExternalizable)); - setWriter(new UdpSocketChannelEntryWriter(serializedEntrySize, entryExternalizable, - modificationIterator, this)); + setWriter(new UdpSocketChannelEntryWriter(replicationConfig.udpBufferSize(), + entryExternalizable, + modificationIterator, this, new Replicators.OutBuffer(UPD_BUFFER_SIZE))); start(); } @@ -121,17 +126,19 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi private final EntryCallback entryCallback; private final UdpChannelReplicator udpReplicator; + private final Replicators.OutBuffer outBuffer; private Replica.ModificationIterator modificationIterator; - UdpSocketChannelEntryWriter(final int serializedEntrySize, + UdpSocketChannelEntryWriter(final int updBufferSize, @NotNull final Replica.EntryExternalizable externalizable, @NotNull final Replica.ModificationIterator modificationIterator, - UdpChannelReplicator udpReplicator) { + UdpChannelReplicator udpReplicator, Replicators.OutBuffer outBuffer) { this.udpReplicator = udpReplicator; + this.outBuffer = outBuffer; // we make the buffer twice as large just to give ourselves headroom - entryCallback = new EntryCallback(externalizable, serializedEntrySize); + entryCallback = new EntryCallback(externalizable, updBufferSize); this.modificationIterator = modificationIterator; } @@ -154,8 +161,8 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi public int writeAll(@NotNull final DatagramChannel socketChannel) throws InterruptedException, IOException { - final ByteBufferBytes in = entryCallback.in(); - final ByteBuffer out = entryCallback.out(); + final ByteBufferBytes in = outBuffer.in(); + final ByteBuffer out = outBuffer.out(); out.clear(); in.clear(); diff --git a/src/main/java/net/openhft/chronicle/map/WriterWithSize.java b/src/main/java/net/openhft/chronicle/map/WriterWithSize.java index 4c946e7dd..9164fd5e5 100644 --- a/src/main/java/net/openhft/chronicle/map/WriterWithSize.java +++ b/src/main/java/net/openhft/chronicle/map/WriterWithSize.java @@ -25,14 +25,19 @@ import net.openhft.lang.threadlocal.Provider; import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.Nullable; +import java.nio.BufferOverflowException; + final class WriterWithSize { private final SizeMarshaller sizeMarshaller; private final Object originalWriter; private final Provider writerProvider; private final MetaBytesInterop originalMetaWriter; private final MetaProvider metaWriterProvider; + private final BufferResizer bufferResizer; - WriterWithSize(SerializationBuilder serializationBuilder) { + WriterWithSize(SerializationBuilder serializationBuilder, + @Nullable BufferResizer bufferResizer) { + this.bufferResizer = bufferResizer; sizeMarshaller = serializationBuilder.sizeMarshaller(); originalWriter = serializationBuilder.interop(); writerProvider = Provider.of((Class) originalWriter.getClass()); @@ -50,11 +55,25 @@ final class WriterWithSize { Object writer = writerProvider.get(copies, originalWriter); copies = metaWriterProvider.getCopies(copies); MetaBytesWriter metaWriter = metaWriterProvider.get(copies, originalMetaWriter, writer, t); - sizeMarshaller.writeSize(out, metaWriter.size(writer, t)); + long size = metaWriter.size(writer, t); + if (bufferResizer != null) + out = resizeIfRequired(out, size); + sizeMarshaller.writeSize(out, size); metaWriter.write(writer, out, t); return copies; } + private Bytes resizeIfRequired(Bytes out, long size) { + if (out.remaining() < size + 8) { + long newCapacity = out.remaining() + size + 8; + if (newCapacity > Integer.MAX_VALUE) + throw new BufferOverflowException(); + return bufferResizer.resizeBuffer((int) newCapacity); + } + return out; + } + + public ThreadLocalCopies writeNullable(Bytes out, T t, @Nullable ThreadLocalCopies copies) { out.writeBoolean(t == null); if (t == null) @@ -71,7 +90,10 @@ final class WriterWithSize { @Nullable ThreadLocalCopies copies) { copies = metaWriterProvider.getCopies(copies); MetaBytesWriter metaWriter = metaWriterProvider.get(copies, originalMetaWriter, writer, t); - sizeMarshaller.writeSize(out, metaWriter.size(writer, t)); + long size = metaWriter.size(writer, t); + if (bufferResizer != null) + out = resizeIfRequired(out, size); + sizeMarshaller.writeSize(out, size); metaWriter.write(writer, out, t); return copies; } @@ -84,3 +106,12 @@ final class WriterWithSize { return writeInLoop(out, t, writer, copies); } } + +interface BufferResizer { + + /** + * @param newCapacity + * @return the newly resize buffer + */ + public Bytes resizeBuffer(int newCapacity); +} diff --git a/src/test/java/net/openhft/chronicle/map/SerializerTest.java b/src/test/java/net/openhft/chronicle/map/SerializerTest.java index ce7d73b28..d8bfd6433 100644 --- a/src/test/java/net/openhft/chronicle/map/SerializerTest.java +++ b/src/test/java/net/openhft/chronicle/map/SerializerTest.java @@ -91,7 +91,7 @@ public class SerializerTest { builder.preMapConstruction(false); - WriterWithSize valueWriterWithSize = new WriterWithSize(builder.valueBuilder); + WriterWithSize valueWriterWithSize = new WriterWithSize(builder.valueBuilder,null); ReaderWithSize valueReaderWithSize = new ReaderWithSize(builder.valueBuilder); valueWriterWithSize.write(out, value, null); @@ -114,7 +114,7 @@ public class SerializerTest { builder.preMapConstruction(false); { - WriterWithSize keyWriterWithSize = new WriterWithSize(builder.keyBuilder); + WriterWithSize keyWriterWithSize = new WriterWithSize(builder.keyBuilder, null); ReaderWithSize keyReaderWithSize = new ReaderWithSize(builder.keyBuilder); keyWriterWithSize.write(out, key, null); @@ -207,7 +207,7 @@ public class SerializerTest { @Override public boolean equals(Object o) { if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; + if (o == null || MyTestClassMarshallable.class != o.getClass()) return false; MyTestClassMarshallable that = (MyTestClassMarshallable) o; From 709e94716630b741224976aacc0cb1ffb49675d7 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Sat, 20 Dec 2014 14:55:41 +0000 Subject: [PATCH 7/7] HCOLL-197 - improved buffer resizing and removed entry size from the stateless client --- .../chronicle/map/StatelessChronicleMap.java | 143 ++++++------------ .../openhft/chronicle/map/WriterWithSize.java | 4 +- 2 files changed, 45 insertions(+), 102 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index f8f4cf2c5..9a973ffc9 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -81,7 +81,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private CloseablesManager closeables; @NotNull private final StatelessMapConfig config; - private int maxEntrySize; + private final Class kClass; private final Class vClass; private final boolean putReturnsNull; @@ -157,21 +157,17 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon this.putReturnsNull = chronicleMapBuilder.putReturnsNull(); this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull(); - this.maxEntrySize = chronicleMapBuilder.entrySize(true); - - if (maxEntrySize < 128) - maxEntrySize = 128; // the stateless client will not work with extreemly small buffers this.vClass = chronicleMapBuilder.valueBuilder.eClass; this.kClass = chronicleMapBuilder.keyBuilder.eClass; this.name = config.name(); attemptConnect(config.remoteAddress()); - outBuffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder()); + outBuffer = allocateDirect(128).order(ByteOrder.nativeOrder()); outBytes = new ByteBufferBytes(outBuffer.slice()); - inBuffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder()); + inBuffer = allocateDirect(128).order(ByteOrder.nativeOrder()); inBytes = new ByteBufferBytes(inBuffer.slice()); } @@ -504,18 +500,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchObject(PUT_MAPPED, key, unaryOperator); } - - private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) { - assert outBytesLock.isHeldByCurrentThread(); - assert !inBytesLock.isHeldByCurrentThread(); - final long remaining = outBytes.remaining(); - - if (remaining < maxEntrySize) { - long estimatedRequiredSize = estimateSize(numberOfEntries, numberOfEntriesReadSoFar); - resizeBufferOutBuffer((int) (estimatedRequiredSize + maxEntrySize), start); - } - } - /** * estimates the size based on what been completed so far */ @@ -756,10 +740,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon 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 + " " + @@ -777,10 +757,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon writeValueInLoop(value, valueWriter, copies); - final int len = (int) (outBytes.position() - start); - - if (len > maxEntrySize) - maxEntrySize = len; } transactionId = send(sizeLocation, startTime); @@ -1306,18 +1282,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long start = outBytes.position(); for (; ; ) { - try { - return keyWriterWithSize.write(outBytes, key, copies); - } catch (IndexOutOfBoundsException e) { - resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start); - } catch (IllegalArgumentException e) { - resizeToMessageOutBuffer(start, e); - } catch (IllegalStateException e) { - if (e.getMessage().contains("Not enough available space")) - resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start); - else - throw e; - } + + return keyWriterWithSize.write(outBytes, key, copies); + } } @@ -1327,22 +1294,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon assert outBytesLock.isHeldByCurrentThread(); assert !inBytesLock.isHeldByCurrentThread(); - long start = outBytes.position(); - for (; ; ) { - try { - return keyWriterWithSize.writeInLoop(outBytes, key, writer, copies); - } catch (IndexOutOfBoundsException e) { - resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start); - } catch (IllegalArgumentException e) { - resizeToMessageOutBuffer(start, e); - } catch (IllegalStateException e) { - if (e.getMessage().contains("Not enough available space")) - resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start); - else - throw e; - } - } + return keyWriterWithSize.writeInLoop(outBytes, key, writer, copies); + } private ThreadLocalCopies writeValue(V value, ThreadLocalCopies copies) { @@ -1350,20 +1304,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon assert !inBytesLock.isHeldByCurrentThread(); long start = outBytes.position(); - for (; ; ) { - try { - assert outBytes.position() == start; - outBytes.limit(outBytes.capacity()); - return valueWriterWithSize.write(outBytes, value, copies); - } catch (IllegalArgumentException e) { - resizeToMessageOutBuffer(start, e); - } catch (IllegalStateException e) { - if (e.getMessage().contains("Not enough available space")) - resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start); - else - throw e; - } - } + + + assert outBytes.position() == start; + outBytes.limit(outBytes.capacity()); + return valueWriterWithSize.write(outBytes, value, copies); + + } @SuppressWarnings("UnusedReturnValue") @@ -1371,40 +1318,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon assert outBytesLock.isHeldByCurrentThread(); assert !inBytesLock.isHeldByCurrentThread(); long start = outBytes.position(); - for (; ; ) { - try { - assert outBytes.position() == start; - outBytes.limit(outBytes.capacity()); - return valueWriterWithSize.writeInLoop(outBytes, value, writer, copies); - } catch (IllegalArgumentException e) { - resizeToMessageOutBuffer(start, e); - } catch (IllegalStateException e) { - if (e.getMessage().contains("Not enough available space")) - resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start); - else - throw e; - } - } - } - private void resizeToMessageOutBuffer(long start, @NotNull IllegalArgumentException e) { - e.printStackTrace(); - assert outBytesLock.isHeldByCurrentThread(); - assert !inBytesLock.isHeldByCurrentThread(); - String message = e.getMessage(); - if (message.startsWith(START_OF)) { - String substring = message.substring("Attempt to write ".length(), message.length()); - int i = substring.indexOf(' '); - if (i != -1) { - int size = Integer.parseInt(substring.substring(0, i)); + assert outBytes.position() == start; + outBytes.limit(outBytes.capacity()); + return valueWriterWithSize.writeInLoop(outBytes, value, writer, copies); - long requiresExtra = size - outBytes.remaining(); - resizeBufferOutBuffer((int) (outBytes.capacity() + requiresExtra), start); - } else - resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start); - } else - throw e; } private V readValue(long transactionId, long startTime, ThreadLocalCopies copies) { @@ -1771,13 +1690,37 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon if (cause instanceof IOException && cause.getMessage().contains("Not enough available space")) { LOG.debug("resizing buffer, name=" + name); - resizeBufferOutBuffer(outBuffer.capacity() + maxEntrySize, start); + + try { + resizeToMessageOutBuffer(start, e); + } catch (Exception e2) { + throw e; + } + } else throw e; } } } + private void resizeToMessageOutBuffer(long start, @NotNull Exception e) throws Exception{ + assert outBytesLock.isHeldByCurrentThread(); + assert !inBytesLock.isHeldByCurrentThread(); + String message = e.getMessage(); + if (message.startsWith("java.io.IOException: Not enough available space for writing ")) { + String substring = message.substring("java.io.IOException: Not enough available space for writing ".length(), message.length()); + int i = substring.indexOf(' '); + if (i != -1) { + int size = Integer.parseInt(substring.substring(0, i)); + + long requiresExtra = size - outBytes.remaining(); + resizeBufferOutBuffer((int) (outBytes.capacity() + requiresExtra), start); + } else + throw e; + } else + throw e; + } + @Nullable private R fetchObject(@NotNull final EventId eventId, K key, @NotNull Object object) { diff --git a/src/main/java/net/openhft/chronicle/map/WriterWithSize.java b/src/main/java/net/openhft/chronicle/map/WriterWithSize.java index 9164fd5e5..d497c2195 100644 --- a/src/main/java/net/openhft/chronicle/map/WriterWithSize.java +++ b/src/main/java/net/openhft/chronicle/map/WriterWithSize.java @@ -65,10 +65,10 @@ final class WriterWithSize { private Bytes resizeIfRequired(Bytes out, long size) { if (out.remaining() < size + 8) { - long newCapacity = out.remaining() + size + 8; + long newCapacity = out.capacity() + (size - out.remaining()) + 8; if (newCapacity > Integer.MAX_VALUE) throw new BufferOverflowException(); - return bufferResizer.resizeBuffer((int) newCapacity); + return bufferResizer.resizeBuffer((int) newCapacity); } return out; }