From 26db7801052f145a0faf724f24f059e71cfa7399 Mon Sep 17 00:00:00 2001 From: danielshaya Date: Wed, 13 May 2015 15:17:58 +0100 Subject: [PATCH] HCOLL-321 Remote implementation for FilePerKeyMap Refactoring --- .../net/openhft/chronicle/map/FPMEvent.java | 13 +- .../chronicle/map/FilePerKeyChronicleMap.java | 52 ++-- .../openhft/chronicle/map/FilePerKeyMap.java | 243 ++++++++++-------- .../map/DeflatorStringMarshaller.java | 143 +++++++++++ .../chronicle/map/FilePerKeyMapTest.java | 13 +- .../chronicle/map/WriteThroughputTest.java | 106 ++++++++ 6 files changed, 429 insertions(+), 141 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/DeflatorStringMarshaller.java create mode 100644 src/test/java/net/openhft/chronicle/map/WriteThroughputTest.java diff --git a/src/main/java/net/openhft/chronicle/map/FPMEvent.java b/src/main/java/net/openhft/chronicle/map/FPMEvent.java index 5f63d05d5..c2153b050 100644 --- a/src/main/java/net/openhft/chronicle/map/FPMEvent.java +++ b/src/main/java/net/openhft/chronicle/map/FPMEvent.java @@ -5,14 +5,12 @@ package net.openhft.chronicle.map; */ public class FPMEvent { private EventType eventType; - private boolean programmatic; private String key; private String value; private String lastValue; - public FPMEvent(EventType eventType, boolean programmatic, String key, String lastValue, String value) { + public FPMEvent(EventType eventType, String key, String lastValue, String value) { this.eventType = eventType; - this.programmatic = programmatic; this.key = key; this.value = value; this.lastValue = lastValue; @@ -34,14 +32,6 @@ public class FPMEvent { this.eventType = eventType; } - public boolean isProgrammatic() { - return programmatic; - } - - public void setProgrammatic(boolean programmatic) { - this.programmatic = programmatic; - } - public String getKey() { return key; } @@ -62,7 +52,6 @@ public class FPMEvent { public String toString() { return "FPMEvent{" + "eventType=" + eventType + - ", programmatic=" + programmatic + ", key='" + key + '\'' + ", value='" + value + '\'' + ", lastValue='" + lastValue + '\'' + diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java index 84cf435e4..5e33da222 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java @@ -1,7 +1,10 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.function.SerializableFunction; +import net.openhft.lang.Jvm; import org.jetbrains.annotations.NotNull; +import org.xerial.snappy.SnappyInputStream; +import org.xerial.snappy.SnappyOutputStream; import java.io.File; import java.io.IOException; @@ -10,18 +13,27 @@ import java.util.Map; import java.util.Set; import java.util.function.Consumer; import java.util.function.Predicate; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; /** * Created by daniel on 22/04/15. */ public class FilePerKeyChronicleMap implements ChronicleMap { - private final ChronicleMap chMap; + // private final ChronicleMap chMap; private final FilePerKeyMap fpMap; - public FilePerKeyChronicleMap(String dir) { + public FilePerKeyChronicleMap(String dir) throws IOException { fpMap = new FilePerKeyMap(dir); - chMap = ChronicleMapBuilder.of(String.class, String.class) - .entries(500).averageValueSize(2_000_000).create(); + fpMap.putReturnsNull(true); +// fpMap.valueMarshaller(SnappyInputStream::new, SnappyOutputStream::new); +// fpMap.valueMarshaller(GZIPInputStream::new, GZIPOutputStream::new); +// chMap = ChronicleMapBuilder.of(String.class, String.class) +// .entries(500).averageValueSize(2_000_000).create(); + } + + public void registerForEvents(Consumer listener) { + fpMap.registerForEvents(listener); } @Override @@ -51,20 +63,20 @@ public class FilePerKeyChronicleMap implements ChronicleMap { @Override public String put(String key, String value) { - fpMap.put(key, value); - return chMap.put(key,value); + return fpMap.put(key, value); +// return chMap.put(key,value); } @Override public String remove(Object key) { - chMap.remove(key); +// chMap.remove(key); return fpMap.remove(key); } @Override public void putAll(Map m) { fpMap.putAll(m); - chMap.putAll(m); +// chMap.putAll(m); } @Override @@ -92,24 +104,28 @@ public class FilePerKeyChronicleMap implements ChronicleMap { @Override public String getUsing(String key, String usingValue) { - return chMap.getUsing(key, usingValue); + throw new UnsupportedOperationException(); +// return chMap.getUsing(key, usingValue); } @Override public String acquireUsing(@NotNull String key, String usingValue) { + throw new UnsupportedOperationException(); //@todo ADD THE LOGIC FOR FPMAP - return chMap.acquireUsing(key, usingValue); +// return chMap.acquireUsing(key, usingValue); } @NotNull @Override public MapKeyContext acquireContext(@NotNull String key, @NotNull String usingValue) { - return chMap.acquireContext(key,usingValue); + throw new UnsupportedOperationException(); +// return chMap.acquireContext(key,usingValue); } @Override public R getMapped(String key, @NotNull SerializableFunction function) { - return chMap.getMapped(key, function); + throw new UnsupportedOperationException(); +// return chMap.getMapped(key, function); } @Override @@ -154,7 +170,8 @@ public class FilePerKeyChronicleMap implements ChronicleMap { @Override public MapKeyContext context(String key) { - return chMap.context(key); + throw new UnsupportedOperationException(); + // return chMap.context(key); } @Override @@ -164,17 +181,20 @@ public class FilePerKeyChronicleMap implements ChronicleMap { @Override public boolean forEachEntryWhile(Predicate> predicate) { - return chMap.forEachEntryWhile(predicate); + throw new UnsupportedOperationException(); + + // return chMap.forEachEntryWhile(predicate); } @Override public void forEachEntry(Consumer> action) { - chMap.forEachEntry(action); + throw new UnsupportedOperationException(); +// chMap.forEachEntry(action); } @Override public void close() { - chMap.close(); + fpMap.close(); } @Override diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java index 851aa905d..211e51c71 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java @@ -1,10 +1,14 @@ package net.openhft.chronicle.map; import com.sun.nio.file.SensitivityWatchEventModifier; +import net.openhft.chronicle.core.util.ThrowingFunction; +import org.jcp.xml.dsig.internal.SignerOutputStream; import org.jetbrains.annotations.NotNull; +import org.xerial.snappy.SnappyInputStream; +import org.xerial.snappy.SnappyOutputStream; -import java.io.File; -import java.io.IOException; +import java.io.*; +import java.nio.charset.StandardCharsets; import java.nio.file.*; import java.util.*; import java.util.concurrent.ConcurrentHashMap; @@ -34,32 +38,48 @@ import java.util.stream.Stream; * very few events if they are done quickly and there is a significant delay * between the event and the event being triggered. */ -public class FilePerKeyMap implements Map { +public class FilePerKeyMap implements Map, Closeable { private final Path dirPath; - private final Map lastModifiedByProgram = new ConcurrentHashMap<>(); - private final Map lastUpdate = new ConcurrentHashMap<>(); - private final List> listeners = new ArrayList<>(); - private final FPMWatcher fileFpmWatcher = new FPMWatcher(); + private final Map lastFile = new ConcurrentHashMap<>(); - public FilePerKeyMap(String dir){ + private final List> listeners = new ArrayList<>(); + private final Thread fileFpmWatcher; + private volatile boolean closed = false; + private boolean putReturnsNull; + private boolean snappyValues; + private ThrowingFunction reading = i -> i; + private ThrowingFunction writing = o -> o; + + public FilePerKeyMap(String dir) throws IOException { this.dirPath = Paths.get(dir); - try { - Files.createDirectories(dirPath); - }catch (IOException e){ - throw new RuntimeException(e); - } + Files.createDirectories(dirPath); + WatchService watcher = FileSystems.getDefault().newWatchService(); + dirPath.register(watcher, new WatchEvent.Kind[]{ + StandardWatchEventKinds.ENTRY_CREATE, + StandardWatchEventKinds.ENTRY_DELETE, + StandardWatchEventKinds.ENTRY_MODIFY}, + SensitivityWatchEventModifier.HIGH + ); + + fileFpmWatcher = new Thread(new FPMWatcher(watcher), dir + "-watcher"); fileFpmWatcher.start(); } - public void registerForEvents(Consumer listener){ + public void registerForEvents(Consumer listener) { listeners.add(listener); } - public void unregisterForEvents(Consumer listener){ + public void unregisterForEvents(Consumer listener) { listeners.remove(listener); } - private void fireEvent(FPMEvent event){ + public void valueMarshaller(ThrowingFunction reading, + ThrowingFunction writing) { + this.reading = reading; + this.writing = writing; + } + + private void fireEvent(FPMEvent event) { for (Consumer listener : listeners) { listener.accept(event); } @@ -67,22 +87,22 @@ public class FilePerKeyMap implements Map { @Override public int size() { - return (int)getFiles().count(); + return (int) getFiles().count(); } @Override public boolean isEmpty() { - return size()==0 ? true: false; + return size() == 0; } @Override public boolean containsKey(Object key) { - return getFiles().anyMatch(p->p.getFileName().toString().equals(key)); + return getFiles().anyMatch(p -> p.getFileName().toString().equals(key)); } @Override public boolean containsValue(Object value) { - return getFiles().anyMatch(p->getFileContents(p).equals(value)); + return getFiles().anyMatch(p -> getFileContents(p).equals(value)); } @Override @@ -93,8 +113,9 @@ public class FilePerKeyMap implements Map { @Override public String put(String key, String value) { + if (closed) throw new IllegalStateException("closed"); Path path = dirPath.resolve(key); - String existingValue = getFileContents(path); + String existingValue = putReturnsNull ? null : getFileContents(path); writeToFile(path, value); return existingValue; @@ -102,8 +123,9 @@ public class FilePerKeyMap implements Map { @Override public String remove(Object key) { + if (closed) throw new IllegalStateException("closed"); String existing = get(key); - if(existing != null){ + if (existing != null) { deleteFile(dirPath.resolve((String) key)); } return existing; @@ -111,7 +133,7 @@ public class FilePerKeyMap implements Map { @Override public void putAll(Map m) { - m.entrySet().stream().forEach(e->put(e.getKey(), e.getValue())); + m.entrySet().stream().forEach(e -> put(e.getKey(), e.getValue())); } @Override @@ -143,8 +165,7 @@ public class FilePerKeyMap implements Map { } - - private Stream getFiles(){ + private Stream getFiles() { try { return Files.walk(dirPath).filter(p -> !Files.isDirectory(p)); } catch (IOException e) { @@ -152,48 +173,64 @@ public class FilePerKeyMap implements Map { } } - private String getFileContents(Path path){ - String existingValue = null; - if(Files.exists(path)) { - try { - existingValue = Files.readAllLines(path).stream().collect(Collectors.joining("\n")); - } catch (IOException e) { - throw new RuntimeException(e); - } + String getFileContents(Path path) { + try { + File file = path.toFile(); + FileRecord last = lastFile.get(file); + if (last != null && file.lastModified() == last.timestamp) + return last.contents; + return getFileContents0(path); + } catch (IOException ioe) { + throw new IllegalStateException(ioe); } - return existingValue; } - private void writeToFile(Path path, String value){ - try { - Files.write(path, value.getBytes(), StandardOpenOption.CREATE, - StandardOpenOption.TRUNCATE_EXISTING, - StandardOpenOption.WRITE); - lastModifiedByProgram.put(path.toFile(), path.toFile().lastModified()); + String getFileContents0(Path path) throws IOException { + if (!Files.exists(path)) return null; + File file = path.toFile(); + ByteArrayOutputStream baos = new ByteArrayOutputStream((int) file.length()); + byte[] bytes = new byte[1024]; + try (InputStream fis = reading.apply(new FileInputStream(file))) { + for (int len; (len = fis.read(bytes)) > 0; ) + baos.write(bytes, 0, len); + } + return baos.toString(); + } + + private void writeToFile(Path path, String value) { + File file = path.toFile(); + File tmpFile = new File(file.getParentFile(), "." + file.getName()); + try (PrintWriter pw = new PrintWriter(new BufferedOutputStream(writing.apply(new FileOutputStream(tmpFile))))) { + pw.write(value); } catch (IOException e) { throw new AssertionError(e); } + tmpFile.renameTo(file); } - private void deleteFile(Path path){ - try { - Files.delete(path); - lastModifiedByProgram.put(path.toFile(), path.toFile().lastModified()); - } catch (IOException e) { - throw new AssertionError(e); - } + private void deleteFile(Path path) { + File key = path.toFile(); + key.delete(); } - public void close(){ + public void close() { + closed = true; fileFpmWatcher.interrupt(); } - private static class FPMEntry implements Entry - { + public void putReturnsNull(boolean putReturnsNull) { + this.putReturnsNull = putReturnsNull; + } + + public void snappyValues(boolean snappyValues) { + this.snappyValues = snappyValues; + } + + private static class FPMEntry implements Entry { private String key; private String value; - public FPMEntry(String key, String value){ + public FPMEntry(String key, String value) { this.key = key; this.value = value; } @@ -234,92 +271,92 @@ public class FilePerKeyMap implements Map { } } - private class FPMWatcher extends Thread{ - @Override - public void run(){ - try { - WatchService watcher = FileSystems.getDefault().newWatchService(); - dirPath.register(watcher, new WatchEvent.Kind[]{ - StandardWatchEventKinds.ENTRY_CREATE, - StandardWatchEventKinds.ENTRY_DELETE, - StandardWatchEventKinds.ENTRY_MODIFY}, - SensitivityWatchEventModifier.HIGH - ); + private class FPMWatcher implements Runnable { + private final WatchService watcher; - while(true){ + public FPMWatcher(WatchService watcher) { + this.watcher = watcher; + } + + @Override + public void run() { + try { + while (true) { WatchKey key = null; try { key = watcher.take(); for (WatchEvent event : key.pollEvents()) { WatchEvent.Kind kind = event.kind(); + if (kind == StandardWatchEventKinds.OVERFLOW) { + // todo log a warning. + continue; + } + // get file name WatchEvent ev = (WatchEvent) event; Path fileName = ev.context(); String mapKey = fileName.toString(); - if(mapKey.startsWith(".")){ + if (mapKey.startsWith(".")) { //this avoids temporary files being added to the map continue; } - if (kind == StandardWatchEventKinds.OVERFLOW) { - continue; - } else if (kind == StandardWatchEventKinds.ENTRY_CREATE) { + if (kind == StandardWatchEventKinds.ENTRY_CREATE) { Path p = dirPath.resolve(fileName); - String mapVal = getFileContents(p); - lastUpdate.put(p.toFile(), mapVal); - if(isProgrammaticUpdate(p.toFile())){ - fireEvent(new FPMEvent(FPMEvent.EventType.NEW, true, mapKey,null, mapVal)); - }else { - fireEvent(new FPMEvent(FPMEvent.EventType.NEW, false, mapKey, null, mapVal)); + try { + String mapVal = getFileContents0(p); + lastFile.put(p.toFile(), new FileRecord(p.toFile().lastModified(), mapVal)); + fireEvent(new FPMEvent(FPMEvent.EventType.NEW, mapKey, null, mapVal)); + } catch (FileNotFoundException ignored) { } } else if (kind == StandardWatchEventKinds.ENTRY_DELETE) { Path p = dirPath.resolve(fileName); - String lastVal = lastUpdate.get(p.toFile()); - lastUpdate.remove(p.toFile()); - - if(isProgrammaticUpdate(p.toFile())){ - fireEvent(new FPMEvent(FPMEvent.EventType.DELETE, true, mapKey, lastVal, null)); - }else { - fireEvent(new FPMEvent(FPMEvent.EventType.DELETE, false, mapKey, lastVal, null)); - } + FileRecord lastVal = lastFile.remove(p.toFile()); + String lastContent = lastVal == null ? null : lastVal.contents; + fireEvent(new FPMEvent(FPMEvent.EventType.DELETE, mapKey, lastContent, null)); } else if (kind == StandardWatchEventKinds.ENTRY_MODIFY) { - Path p = dirPath.resolve(fileName); - String mapVal = getFileContents(p); - String lastVal = null; - if(mapVal != null) { - lastVal = lastUpdate.put(p.toFile(), mapVal); - } + try { + Path p = dirPath.resolve(fileName); + String mapVal = getFileContents0(p); + String lastVal = null; + if (mapVal != null) { + FileRecord rec = lastFile.put(p.toFile(), new FileRecord(p.toFile().lastModified(), mapVal)); + if (rec != null) lastVal = rec.contents; + } - if(lastVal != null && lastVal.equals(mapVal)){ - //Nothing has changed don't fire an event - continue; - } + if (lastVal != null && lastVal.equals(mapVal)) { + //Nothing has changed don't fire an event + continue; + } - if(isProgrammaticUpdate(p.toFile())){ - fireEvent(new FPMEvent(FPMEvent.EventType.UPDATE, true, mapKey,lastVal,mapVal)); - }else { - fireEvent(new FPMEvent(FPMEvent.EventType.UPDATE, false, mapKey,lastVal,mapVal)); + fireEvent(new FPMEvent(FPMEvent.EventType.UPDATE, mapKey, lastVal, mapVal)); + } catch (FileNotFoundException ignored) { } } } } catch (InterruptedException e) { return; - }finally{ - if(key!=null)key.reset(); + } finally { + if (key != null) key.reset(); } } - } catch (IOException e) { - throw new RuntimeException(e); + } catch (Throwable e) { + if (!closed) + e.printStackTrace(); } } } +} - private boolean isProgrammaticUpdate(File file) { - return lastModifiedByProgram.containsKey(file) - && (file.lastModified() == lastModifiedByProgram.get(file)); +class FileRecord { + final long timestamp; + final String contents; + + FileRecord(long timestamp, String contents) { + this.timestamp = timestamp; + this.contents = contents; } - } diff --git a/src/test/java/net/openhft/chronicle/map/DeflatorStringMarshaller.java b/src/test/java/net/openhft/chronicle/map/DeflatorStringMarshaller.java new file mode 100644 index 000000000..795aba23c --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/DeflatorStringMarshaller.java @@ -0,0 +1,143 @@ +package net.openhft.chronicle.map; + +import net.openhft.chronicle.bytes.IORuntimeException; +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.CompactBytesMarshaller; +import org.jetbrains.annotations.Nullable; + +import java.io.*; +import java.lang.reflect.Constructor; +import java.nio.ByteBuffer; +import java.util.zip.DeflaterOutputStream; +import java.util.zip.InflaterInputStream; + +/** + * Created by peter on 24/10/14. + */ +public enum DeflatorStringMarshaller implements CompactBytesMarshaller { + INSTANCE; + + private static final StringFactory STRING_FACTORY = getStringFactory(); + + private static final int NULL_LENGTH = -1; + + private static StringFactory getStringFactory() { + try { + return new StringFactory17(); + } catch (Exception e) { + // do nothing + } + + try { + return new StringFactory16(); + } catch (Exception e) { + // no more alternatives + throw new AssertionError(e); + } + } + + @Override + public byte code() { + return '_'; + } + + @Override + public void write(Bytes bytes, CharSequence s) { + if (s == null) { + bytes.writeStopBit(NULL_LENGTH); + return; + } else if (s.length() == 0) { + bytes.writeStopBit(0); + return; + } + // write the total length. + int length = s.length(); + bytes.writeStopBit(length); + + long position = bytes.position(); + bytes.writeInt(0); + DataOutputStream dos = new DataOutputStream(new BufferedOutputStream( + new DeflaterOutputStream(bytes.outputStream()))); + try { + for (int i = 0; i < s.length(); i++) { + dos.write(s.charAt(i)); + } + dos.close(); + } catch (IOException e) { + throw new IORuntimeException(e); + } + bytes.writeInt(position, (int) (bytes.position() - position - 4)); + } + + @Override + public String read(Bytes bytes) { + return read(bytes, null); + } + + @Override + public String read(Bytes bytes, @Nullable CharSequence ignored) { + long size = bytes.readStopBit(); + if (size == NULL_LENGTH) + return null; + if (size < 0 || size > Integer.MAX_VALUE) + throw new IllegalStateException("Invalid length: " + size); + if (size == 0) + return ""; + + char[] chars = new char[(int) size]; + int length = bytes.readInt(); + long limit = bytes.limit(); + try { + bytes.limit(bytes.position() + length); + DataInputStream dis = new DataInputStream(new BufferedInputStream( + new InflaterInputStream(bytes.inputStream()))); + for (int i = 0; i < size; i++) + chars[i] = (char) (dis.readByte() & 0xff); + + } catch (IOException e) { + throw new IORuntimeException(e); + } finally { + bytes.position(bytes.limit()); + bytes.limit(limit); + } + try { + return STRING_FACTORY.fromChars(chars); + } catch (Exception e) { + throw new AssertionError(e); + } + } + + private static abstract class StringFactory { + abstract String fromChars(char[] chars) throws Exception; + } + + private static final class StringFactory16 extends StringFactory { + private final Constructor constructor; + + private StringFactory16() throws Exception { + constructor = String.class.getDeclaredConstructor(int.class, + int.class, char[].class); + constructor.setAccessible(true); + } + + @Override + String fromChars(char[] chars) throws Exception { + return constructor.newInstance(0, chars.length, chars); + } + } + + private static final class StringFactory17 extends StringFactory { + private final Constructor constructor; + + private StringFactory17() throws Exception { + constructor = String.class.getDeclaredConstructor(char[].class, boolean.class); + constructor.setAccessible(true); + } + + @Override + String fromChars(char[] chars) throws Exception { + return constructor.newInstance(chars, true); + } + } +} diff --git a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java index 21279d6ba..b8a919b74 100644 --- a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java @@ -4,6 +4,7 @@ import org.junit.BeforeClass; import org.junit.Ignore; import org.junit.Test; +import java.io.IOException; import java.util.*; import java.util.function.Consumer; @@ -15,7 +16,7 @@ import static org.junit.Assert.*; public class FilePerKeyMapTest { private static FilePerKeyMap map; @BeforeClass - public static void createMap(){ + public static void createMap() throws IOException { map = new FilePerKeyMap("/tmp/filepermaptests"); //just in case it hasn't been cleared up last time map.clear(); @@ -133,7 +134,7 @@ public class FilePerKeyMapTest { @Ignore //This doesn't work on all OS @Test - public void eventTest() { + public void eventTest() throws InterruptedException { final List events = new ArrayList<>(); Consumer fpmEventConsumer = (FPMEvent e) -> { @@ -147,7 +148,6 @@ public class FilePerKeyMapTest { map.remove("one"); map.put("one", "one"); - try { //Allow all events to be played through Thread.sleep(3000); //Check the events @@ -158,13 +158,6 @@ public class FilePerKeyMapTest { assertEquals(FPMEvent.EventType.DELETE, events.get(2).getEventType()); assertEquals(FPMEvent.EventType.NEW, events.get(3).getEventType()); - assertEquals(true, events.get(0).isProgrammatic()); - assertEquals(true, events.get(1).isProgrammatic()); - assertEquals(true, events.get(2).isProgrammatic()); - assertEquals(true, events.get(3).isProgrammatic()); - } catch (InterruptedException e) { - e.printStackTrace(); - } map.unregisterForEvents(fpmEventConsumer); map.clear(); diff --git a/src/test/java/net/openhft/chronicle/map/WriteThroughputTest.java b/src/test/java/net/openhft/chronicle/map/WriteThroughputTest.java new file mode 100644 index 000000000..9ab9baf8d --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/WriteThroughputTest.java @@ -0,0 +1,106 @@ +package net.openhft.chronicle.map; + +import net.openhft.lang.io.serialization.impl.SnappyStringMarshaller; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.util.stream.IntStream; + +/** + * Created by peter on 12/05/15. + */ +public class WriteThroughputTest { + @Ignore("Long running") + @Test + public void bandwidthTest() throws IOException { + int count = 2000; + int size = 50 << 10; + + try (ChronicleMap map = ChronicleMapBuilder + .of(CharSequence.class, CharSequence.class) + .entries(count * 3 / 2) + .averageValueSize(size) + .putReturnsNull(true) + .create()) { + + System.out.println("ChronicleMap.put to memory"); + doTest(count, size, map); + } + + File file = new File("bandwidthTest" + System.nanoTime() + ".deleteme"); + file.deleteOnExit(); + try (ChronicleMap map = ChronicleMapBuilder + .of(CharSequence.class, CharSequence.class) + .entries(count * 3 / 2) + .averageValueSize(size) + .putReturnsNull(true) + .createPersistedTo(file)) { + + System.out.println("ChronicleMap.put to disk"); + doTest(count, size, map); + } + + File file2 = new File("bandwidthTest" + System.nanoTime() + ".snappy.deleteme"); + file2.deleteOnExit(); + try (ChronicleMap map = ChronicleMapBuilder + .of(CharSequence.class, CharSequence.class) + .entries(count * 3 / 2) + .averageValueSize(size / 4) + .valueMarshaller(SnappyStringMarshaller.INSTANCE) + .putReturnsNull(true) + .createPersistedTo(file2)) { + + System.out.println("ChronicleMap.put to snappy disk"); + doTest(count, size, map); + } + } + @Ignore("TODO") + @Test + public void bandwidthTestZ() throws IOException { + int count = 2 << 10; + int size = 50 << 10; + + File file3 = new File("bandwidthTest"+System.nanoTime()+ ".Z.deleteme"); + file3.deleteOnExit(); + try (ChronicleMap map = ChronicleMapBuilder + .of(CharSequence.class, CharSequence.class) + .entries(count*3/2) + .averageValueSize(size/4) + .valueMarshaller(DeflatorStringMarshaller.INSTANCE) + .putReturnsNull(true) + .createPersistedTo(file3)) { + + System.out.println("ChronicleMap.put to deflator disk"); + doTest(count, size, map); + } + } + + private void doTest(int count, int size,ChronicleMap map) { + StringBuilder value = new StringBuilder("value"); + while (value.length() < size) + value.append(value.length()); + + for (int t = 0; t < 1; t++) { + StringBuilder key = new StringBuilder("key"); + long start = System.nanoTime(); + for (int i = 0; i < count; i++) + map.put(key, value); + long time = System.nanoTime() - start; + System.out.printf("Serial %,d MB took %.3f seconds%n", size * count >> 20, time / 1e9); + } + + for (int t = 0; t < 3; t++) { + long start = System.nanoTime(); + IntStream.range(0, count).parallel().forEach(i -> { + StringBuilder key = new StringBuilder("key"); + key.append(i); + map.put(key, value); + }); + long time = System.nanoTime() - start; + System.out.printf("Concurrent %,d MB took %.3f seconds%n", size * count >> 20, time / 1e9); + } + + } +}