From 356e6c8181d0d6baf9f76a6b9dfc7a4e1112c275 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Thu, 11 Dec 2014 07:46:05 +0000 Subject: [PATCH] Improve support for large entries --- .../chronicle/map/LargeEntriesTest.java | 112 +++++++++++++----- 1 file changed, 85 insertions(+), 27 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java b/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java index 20421483f..955e9c1b9 100644 --- a/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java +++ b/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java @@ -41,23 +41,23 @@ import static org.junit.Assert.*; */ public class LargeEntriesTest { - static final int ENTRIES = 6400; - static final int ENTRY_SIZE = 102400; - private static final boolean TESTING = Boolean.getBoolean("testing"); - @Test - @Ignore public void testLargeStrings() throws ExecutionException, InterruptedException, IOException { + final int ENTRIES = 250; + final int ENTRY_SIZE = 100 * 1024; + File file = File.createTempFile("largeEntries", ".deleteme"); + file.deleteOnExit(); final ChronicleMap map = ChronicleMapBuilder .of(String.class, String.class) .valueMarshaller(SnappyStringMarshaller.INSTANCE) .actualSegments(1) // to force an error. - .entries(ENTRIES * 2) - .entrySize(ENTRY_SIZE / 4) + .entries(ENTRIES) + .entrySize(ENTRY_SIZE / 6) .putReturnsNull(true) .createPersistedTo(file); { + warmUpCompression(ENTRY_SIZE); int threads = 4; //Runtime.getRuntime().availableProcessors(); ExecutorService es = Executors.newFixedThreadPool(threads); final int block = ENTRIES / threads; @@ -66,11 +66,10 @@ public class LargeEntriesTest { List> futureList = new ArrayList<>(); for (int t = 0; t < threads; t++) { final int finalT = t; - final int finalI = i; futureList.add(es.submit(new Runnable() { @Override public void run() { - exerciseLargeStrings(map, finalI, finalT * block, finalT * block + block, ENTRY_SIZE); + exerciseLargeStrings(map, finalT * block, finalT * block + block, ENTRY_SIZE); } })); } @@ -87,7 +86,70 @@ public class LargeEntriesTest { } } - void exerciseLargeStrings(ChronicleMap map, int run, int start, int finish, int entrySize) { + private void warmUpCompression(int entrySize) { + String value = generateValue(entrySize); + DirectBytes bytes = DirectStore.allocate(entrySize / 6).bytes(); + for (int i = 0; i < 5; i++) { + // warmup to compression. + bytes.clear(); + SnappyStringMarshaller.INSTANCE.write(bytes, value); + bytes.flip(); + SnappyStringMarshaller.INSTANCE.read(bytes); + } + bytes.release(); + } + + @Test + @Ignore + public void testLargeStringsPerf() throws ExecutionException, InterruptedException, IOException { + doLargeEntryPerf(10000, 100 * 1024); + doLargeEntryPerf(100000, 10 * 1024); + doLargeEntryPerf(10000, 100 * 1024); + doLargeEntryPerf(3000, 1024 * 1024); + } + + private void doLargeEntryPerf(int ENTRIES, final int ENTRY_SIZE) throws IOException, InterruptedException, ExecutionException { + System.out.printf("Testing %,d entries of %,d KB%n", ENTRIES, ENTRY_SIZE); + File file = File.createTempFile("largeEntries", ".deleteme"); + file.deleteOnExit(); + final ChronicleMap map = ChronicleMapBuilder + .of(String.class, String.class) + .valueMarshaller(SnappyStringMarshaller.INSTANCE) + .entries(ENTRIES * 2) + .entrySize(ENTRY_SIZE / 6) + .putReturnsNull(true) + .createPersistedTo(file); + { + warmUpCompression(ENTRY_SIZE); + int threads = Runtime.getRuntime().availableProcessors(); + ExecutorService es = Executors.newFixedThreadPool(threads); + final int block = ENTRIES / threads; + for (int i = 0; i < 3; i++) { + long start = System.currentTimeMillis(); + List> futureList = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + final int finalT = t; + futureList.add(es.submit(new Runnable() { + @Override + public void run() { + exerciseLargeStrings(map, finalT * block, finalT * block + block, ENTRY_SIZE); + } + })); + } + for (Future future : futureList) { + future.get(); + } + long time = System.currentTimeMillis() - start; + long operations = 3; + System.out.printf("Put/Get rate was %.1f MB/s%n", operations * ENTRIES * ENTRY_SIZE / 1e6 / (time / 1e3)); + } + es.shutdown(); + if (es.isTerminated()) + map.close(); + } + } + + void exerciseLargeStrings(ChronicleMap map, int start, int finish, int entrySize) { /* final Thread thisThread = Thread.currentThread(); Thread monitor = new Thread(new Runnable() { @@ -110,23 +172,7 @@ public class LargeEntriesTest { }); monitor.start(); */ - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - XMLEncoder xml = new XMLEncoder(baos); - Map map2 = new HashMap<>(); - Random rand = new Random(1); - for (int i = 0; i < entrySize / 80; i++) - map2.put("key-" + i, "value-" + rand.nextInt(1000)); - xml.writeObject(map2); - xml.close(); - String value = baos.toString().substring(0, entrySize); - // warmup to compression. - if (run == 0) { - DirectBytes bytes = DirectStore.allocate(entrySize).bytes(); - SnappyStringMarshaller.INSTANCE.write(bytes, value); - bytes.flip(); - SnappyStringMarshaller.INSTANCE.read(bytes); - bytes.release(); - } + String value = generateValue(entrySize); for (int i = start; i < finish; i++) { String key = "key-" + i; @@ -151,4 +197,16 @@ public class LargeEntriesTest { assertEquals(key, entrySize, object.length()); } } + + private String generateValue(int entrySize) { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + XMLEncoder xml = new XMLEncoder(baos); + Map map2 = new HashMap<>(); + Random rand = new Random(1); + for (int i = 0; i < entrySize / 80; i++) + map2.put("key-" + i, "value-" + rand.nextInt(1000)); + xml.writeObject(map2); + xml.close(); + return baos.toString().substring(0, entrySize); + } }