From e9f995379d80ea5286ea1c7aa804c3d76b3592a9 Mon Sep 17 00:00:00 2001 From: danielshaya Date: Thu, 23 Apr 2015 18:37:53 +0100 Subject: [PATCH 1/8] Initial commit for FilePerMap implementation. --- .../net/openhft/chronicle/map/FPMEvent.java | 62 +++ .../openhft/chronicle/map/FPMListener.java | 10 + .../chronicle/map/FilePerKeyChronicleMap.java | 199 +++++++++ .../openhft/chronicle/map/FilePerKeyMap.java | 384 +++++++----------- .../chronicle/map/FilePerKeyMapTest.java | 43 +- 5 files changed, 440 insertions(+), 258 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/FPMEvent.java create mode 100644 src/main/java/net/openhft/chronicle/map/FPMListener.java create mode 100644 src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java diff --git a/src/main/java/net/openhft/chronicle/map/FPMEvent.java b/src/main/java/net/openhft/chronicle/map/FPMEvent.java new file mode 100644 index 000000000..bf7278e83 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/FPMEvent.java @@ -0,0 +1,62 @@ +package net.openhft.chronicle.map; + +/** + * Created by daniel on 23/04/15. + */ +public class FPMEvent { + public enum EventType {NEW, UPDATE, DELETE} + + private EventType eventType; + private boolean programmatic; + private String key; + private String value; + + public FPMEvent(EventType eventType, boolean programmatic, String key, String value) { + this.eventType = eventType; + this.programmatic = programmatic; + this.key = key; + this.value = value; + } + + public EventType getEventType() { + return eventType; + } + + public void setEventType(EventType eventType) { + this.eventType = eventType; + } + + public boolean isProgrammatic() { + return programmatic; + } + + public void setProgrammatic(boolean programmatic) { + this.programmatic = programmatic; + } + + public String getKey() { + return key; + } + + public void setKey(String key) { + this.key = key; + } + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = value; + } + + @Override + public String toString() { + return "FPMEvent{" + + "eventType=" + eventType + + ", programmatic=" + programmatic + + ", key='" + key + '\'' + + ", value='" + value + '\'' + + '}'; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/FPMListener.java b/src/main/java/net/openhft/chronicle/map/FPMListener.java new file mode 100644 index 000000000..7fba5460b --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/FPMListener.java @@ -0,0 +1,10 @@ +package net.openhft.chronicle.map; + +/** + * Created by daniel on 23/04/15. + */ +public class FPMListener { + public void onEvent(FPMEvent event){ + + } +} diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java new file mode 100644 index 000000000..6648c4c90 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java @@ -0,0 +1,199 @@ +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.function.SerializableFunction; +import org.jetbrains.annotations.NotNull; + +import java.io.File; +import java.io.IOException; +import java.util.Collection; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; +import java.util.function.Predicate; + +/** + * Created by daniel on 22/04/15. + */ +public class FilePerKeyChronicleMap implements ChronicleMap { + private final String dir; + private final Map lastModifiedByProgram = new ConcurrentHashMap<>(); + private final ChronicleMap chMap; + private final FilePerKeyMap fpMap; + + public FilePerKeyChronicleMap(String dir) { + this.dir = dir; + fpMap = new FilePerKeyMap(dir); + chMap = ChronicleMapBuilder.of(String.class, String.class) + .entries(500).averageValueSize(2_000_000).create(); + } + + @Override + public int size() { + return fpMap.size(); + } + + @Override + public boolean isEmpty() { + return fpMap.isEmpty(); + } + + @Override + public boolean containsKey(Object key) { + return fpMap.containsKey(key); + } + + @Override + public boolean containsValue(Object value) { + return fpMap.containsValue(value); + } + + @Override + public String get(Object key) { + return fpMap.get(key); + } + + @Override + public String put(String key, String value) { + return fpMap.put(key, value); + } + + @Override + public String remove(Object key) { + return fpMap.remove(key); + } + + @Override + public void putAll(Map m) { + fpMap.putAll(m); + } + + @Override + public void clear() { + fpMap.clear(); + } + + @NotNull + @Override + public Set keySet() { + return fpMap.keySet(); + } + + @NotNull + @Override + public Collection values() { + return fpMap.values(); + } + + @NotNull + @Override + public Set> entrySet() { + return fpMap.entrySet(); + } + + @Override + public String getUsing(String key, String usingValue) { + throw new UnsupportedOperationException(); + } + + @Override + public String acquireUsing(@NotNull String key, String usingValue) { + throw new UnsupportedOperationException(); + } + + @NotNull + @Override + public MapKeyContext acquireContext(@NotNull String key, @NotNull String usingValue) { + throw new UnsupportedOperationException(); + } + + @Override + public R getMapped( String key, @NotNull SerializableFunction function) { + throw new UnsupportedOperationException(); + } + + @Override + public String putMapped( String key, @NotNull UnaryOperator unaryOperator) { + throw new UnsupportedOperationException(); + } + + @Override + public void getAll(File toFile) throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public void putAll(File fromFile) throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public String newValueInstance() { + return null; + } + + @Override + public String newKeyInstance() { + throw new UnsupportedOperationException(); + } + + @Override + public Class valueClass() { + throw new UnsupportedOperationException(); + } + + @Override + public File file() { + throw new UnsupportedOperationException(); + } + + @Override + public long longSize() { + return size(); + } + + @Override + public MapKeyContext context( String key) { + throw new UnsupportedOperationException(); + } + + @Override + public Class keyClass() { + throw new UnsupportedOperationException(); + } + + @Override + public boolean forEachEntryWhile(Predicate> predicate) { + throw new UnsupportedOperationException(); + } + + @Override + public void forEachEntry(Consumer> action) { + throw new UnsupportedOperationException(); + } + + @Override + public void close() { + throw new UnsupportedOperationException(); + } + + @Override + public String putIfAbsent(String key, String value) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean remove(Object key, Object value) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean replace(String key, String oldValue, String newValue) { + throw new UnsupportedOperationException(); + } + + @Override + public String replace(String key, String value) { + throw new UnsupportedOperationException(); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java index 20ca46293..b94d7ab71 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java @@ -1,7 +1,6 @@ package net.openhft.chronicle.map; import com.sun.nio.file.SensitivityWatchEventModifier; -import net.openhft.chronicle.hash.function.SerializableFunction; import org.jetbrains.annotations.NotNull; import java.io.File; @@ -9,113 +8,42 @@ import java.io.IOException; import java.nio.file.*; import java.util.*; import java.util.concurrent.ConcurrentHashMap; -import java.util.function.Consumer; -import java.util.function.Predicate; import java.util.stream.Collectors; import java.util.stream.Stream; /** - * Created by daniel on 22/04/15. + * */ -public class FilePerKeyMap implements ChronicleMap { - private String dir; - private Path dirPath; - private Map lastModifiedByProgram = new ConcurrentHashMap<>(); +public class FilePerKeyMap implements Map { + private final String dir; + private final Path dirPath; + private final Map lastModifiedByProgram = new ConcurrentHashMap<>(); + private final List listeners = new ArrayList<>(); - public FilePerKeyMap(String dir) { + public FilePerKeyMap(String dir){ + this.dir = dir; + this.dirPath = Paths.get(dir); try { - this.dir = dir; - dirPath = Paths.get(dir); Files.createDirectories(dirPath); - - new Thread("WatcherThread") { - 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 - ); - - while(true){ - WatchKey key = null; - try { - key = watcher.take(); - System.out.println("Event Taken"); - for (WatchEvent event : key.pollEvents()) { - // get event type - WatchEvent.Kind kind = event.kind(); - - // get file name - @SuppressWarnings("unchecked") - WatchEvent ev = (WatchEvent) event; - Path fileName = ev.context(); - - String mapKey = fileName.toString(); - - - if (kind == StandardWatchEventKinds.OVERFLOW) { - System.out.println("OVERFLOW"); - continue; - } else if (kind == StandardWatchEventKinds.ENTRY_CREATE) { - Path p = dirPath.resolve(fileName); - String mapVal = getFileContents(p); - if(isProgrammaticUpdate(p.toFile())){ - System.out.println("Programmatic create to " + mapKey);// + " : " + mapVal); - }else { - System.out.println("created " + mapKey);// + " : " + mapVal); - } - } else if (kind == StandardWatchEventKinds.ENTRY_DELETE) { - Path p = dirPath.resolve(fileName); - if(isProgrammaticUpdate(p.toFile())){ - System.out.println("Programmatic delete to " + mapKey); - lastModifiedByProgram.remove(p.toFile()); - }else { - System.out.println("User delete " + mapKey); - } - } else if (kind == StandardWatchEventKinds.ENTRY_MODIFY) { - Path p = dirPath.resolve(fileName); - String mapVal = getFileContents(p); - if(isProgrammaticUpdate(p.toFile())){ - System.out.println("Programmatic update to " + mapKey);// + " : " + mapVal); - }else { - System.out.println("User modified " + mapKey + " : ");// + mapVal); - } - } - } - - } catch (InterruptedException e) { - System.out.println("Watcher service exiting"); - return; - }finally{ - key.reset(); - } - } - } catch (IOException e) { - throw new AssertionError(e); - } - } - }.start(); - - - } catch (IOException e) { - throw new AssertionError(e); + }catch (IOException e){ + throw new RuntimeException(e); } + new FPMWatcher().start(); } - private boolean isProgrammaticUpdate(File file){ - if(lastModifiedByProgram.containsKey(file)){ - return file.lastModified() == lastModifiedByProgram.get(file) ? true : false; - }else{ - return false; + public void registerForEvents(FPMListener listener){ + listeners.add(listener); + } + + private void fireEvent(FPMEvent event){ + for (FPMListener listener : listeners) { + listener.onEvent(event); } } @Override public int size() { - return (int)getFiles(dirPath).count(); + return (int)getFiles().count(); } @Override @@ -125,184 +53,78 @@ public class FilePerKeyMap implements ChronicleMap { @Override public boolean containsKey(Object key) { - return getFiles(dirPath).anyMatch(p->p.getFileName().toString().equals(key)); + + return getFiles().anyMatch(p->p.getFileName().toString().equals(key)); } @Override public boolean containsValue(Object value) { - return getFiles(dirPath).anyMatch(p->getFileContents(p).equals(value)); + return getFiles().anyMatch(p->getFileContents(p).equals(value)); } @Override - public V get(Object key) { - checkTypeIsString(key); - - Path path = Paths.get(dir,(String)key); - return (V)getFileContents(path); + public String get(Object key) { + Path path = dirPath.resolve((String) key); + return getFileContents(path); } @Override - public V put(K key, V value) { - checkTypeIsString(key, value); - - Path path = Paths.get(dir,(String)key); + public String put(String key, String value) { + Path path = dirPath.resolve(key); String existingValue = getFileContents(path); - writeToFile(path, (String) value); + writeToFile(path, value); - return (V) existingValue; + return existingValue; } @Override - public V remove(Object key) { - String existing = (String)get(key); + public String remove(Object key) { + String existing = get(key); if(existing != null){ deleteFile(Paths.get(dir, (String)key)); } - return (V)existing; + return existing; + } @Override - public void putAll(Map m) { + public void putAll(Map m) { m.entrySet().stream().forEach(e->put(e.getKey(), e.getValue())); } @Override public void clear() { - getFiles(dirPath).forEach(this::deleteFile); + getFiles().forEach(this::deleteFile); } @NotNull @Override - public Set keySet() { - return getFiles(dirPath).map(p -> (K) p.getFileName().toString()) - .collect(Collectors.toSet()); - } - - @NotNull - @Override - public Collection values() { - return getFiles(dirPath).map(p -> (V) getFileContents(p)) + public Set keySet() { + return getFiles().map(p -> p.getFileName().toString()) .collect(Collectors.toSet()); } @NotNull @Override - public Set> entrySet() { - return getFiles(dirPath).map(p -> - (Entry) new FPMEntry(p.getFileName().toString(), getFileContents(p))) + public Collection values() { + return getFiles().map(p -> getFileContents(p)) .collect(Collectors.toSet()); - } - @Override - public V getUsing(K key, V usingValue) { - return null; - } - - @Override - public V acquireUsing(@NotNull K key, V usingValue) { - return null; } @NotNull @Override - public MapKeyContext acquireContext(@NotNull K key, @NotNull V usingValue) { - return null; - } - - @Override - public R getMapped(K key, @NotNull SerializableFunction function) { - return null; - } - - @Override - public V putMapped(K key, @NotNull UnaryOperator unaryOperator) { - return null; - } - - @Override - public void getAll(File toFile) throws IOException { - - } - - @Override - public void putAll(File fromFile) throws IOException { - - } - - @Override - public V newValueInstance() { - return null; - } - - @Override - public K newKeyInstance() { - return null; - } - - @Override - public Class valueClass() { - return null; - } - - @Override - public File file() { - return null; - } - - @Override - public long longSize() { - return 0; - } - - @Override - public MapKeyContext context(K key) { - return null; - } - - @Override - public Class keyClass() { - return null; - } - - @Override - public boolean forEachEntryWhile(Predicate> predicate) { - return false; - } - - @Override - public void forEachEntry(Consumer> action) { - - } - - @Override - public void close() { - - } - - @Override - public V putIfAbsent(K key, V value) { - return null; - } - - @Override - public boolean remove(Object key, Object value) { - return false; - } - - @Override - public boolean replace(K key, V oldValue, V newValue) { - return false; - } - - @Override - public V replace(K key, V value) { - return null; + public Set> entrySet() { + return getFiles().map(p -> + (Entry) new FPMEntry(p.getFileName().toString(), getFileContents(p))) + .collect(Collectors.toSet()); } - private Stream getFiles(Path path){ + + private Stream getFiles(){ try { - return Files.walk(path).filter(p -> !Files.isDirectory(p)); + return Files.walk(dirPath).filter(p -> !Files.isDirectory(p)); } catch (IOException e) { throw new AssertionError(e); } @@ -341,38 +163,31 @@ public class FilePerKeyMap implements ChronicleMap { } - - private void checkTypeIsString(Object... objs) { - for(Object o : objs){ - if(!(o instanceof String)) - throw new AssertionError("FilePerKeyMap only accepts Key and Values " + - "of type String. Unsupported type: '" + o + "'"); - } - } - - private static class FPMEntry implements Entry + private static class FPMEntry implements Entry { - private K key; - private V value; + private String key; + private String value; - public FPMEntry(K key, V value){ + public FPMEntry(String key, String value){ this.key = key; this.value = value; } @Override - public K getKey() { + public String getKey() { return key; } @Override - public V getValue() { + public String getValue() { return value; } @Override - public Object setValue(Object value) { - throw new UnsupportedOperationException(); + public String setValue(String value) { + String lastValue = this.value; + this.value = value; + return lastValue; } @Override @@ -380,10 +195,10 @@ public class FilePerKeyMap implements ChronicleMap { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; - FPMEntry entry = (FPMEntry) o; + FPMEntry fpmEntry = (FPMEntry) o; - if (key != null ? !key.equals(entry.key) : entry.key != null) return false; - return !(value != null ? !value.equals(entry.value) : entry.value != null); + if (key != null ? !key.equals(fpmEntry.key) : fpmEntry.key != null) return false; + return !(value != null ? !value.equals(fpmEntry.value) : fpmEntry.value != null); } @@ -394,4 +209,83 @@ public class FilePerKeyMap implements ChronicleMap { return result; } } + + 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 + ); + + while(true){ + WatchKey key = null; + try { + key = watcher.take(); + for (WatchEvent event : key.pollEvents()) { + // get event type + WatchEvent.Kind kind = event.kind(); + + // get file name + WatchEvent ev = (WatchEvent) event; + Path fileName = ev.context(); + + String mapKey = fileName.toString(); + + + if (kind == StandardWatchEventKinds.OVERFLOW) { + System.out.println("OVERFLOW"); + continue; + } else if (kind == StandardWatchEventKinds.ENTRY_CREATE) { + Path p = dirPath.resolve(fileName); + String mapVal = getFileContents(p); + if(isProgrammaticUpdate(p.toFile())){ + fireEvent(new FPMEvent(FPMEvent.EventType.NEW, true, mapKey,mapVal)); + }else { + fireEvent(new FPMEvent(FPMEvent.EventType.NEW, false, mapKey, mapVal)); + } + } else if (kind == StandardWatchEventKinds.ENTRY_DELETE) { + Path p = dirPath.resolve(fileName); + if(isProgrammaticUpdate(p.toFile())){ + fireEvent(new FPMEvent(FPMEvent.EventType.DELETE, true, mapKey, null)); + lastModifiedByProgram.remove(p.toFile()); + }else { + fireEvent(new FPMEvent(FPMEvent.EventType.DELETE, false, mapKey, null)); + } + } else if (kind == StandardWatchEventKinds.ENTRY_MODIFY) { + Path p = dirPath.resolve(fileName); + String mapVal = getFileContents(p); + if(isProgrammaticUpdate(p.toFile())){ + fireEvent(new FPMEvent(FPMEvent.EventType.UPDATE, true, mapKey,mapVal)); + }else { + fireEvent(new FPMEvent(FPMEvent.EventType.UPDATE, false, mapKey, mapVal)); + } + } + } + + } catch (InterruptedException e) { + System.out.println("Watcher service exiting"); + return; + }finally{ + key.reset(); + } + } + } catch (IOException e) { + throw new AssertionError(e); + } + } + } + + private boolean isProgrammaticUpdate(File file){ + if(lastModifiedByProgram.containsKey(file)){ + return file.lastModified() == lastModifiedByProgram.get(file) ? true : false; + }else{ + return false; + } + } + } diff --git a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java index cab4a770c..fa3590ea3 100644 --- a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java @@ -13,22 +13,13 @@ import static org.junit.Assert.*; * JUnit test class to support {@link FilePerKeyMapTest} */ public class FilePerKeyMapTest { - /** - * FilePerMapKey only works with Strings - */ - @Test(expected = AssertionError.class) - public void testWrongType(){ - FilePerKeyMap map = new FilePerKeyMap<>("/tmp/filepermaptests"); - map.put(1, "test"); - } - /** * Testing all the methods of the map with simple tests. */ @Test public void testMapMethods() { //There are no entries in the map so null should be returned - FilePerKeyMap map = new FilePerKeyMap<>("/tmp/filepermaptests"); + FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests"); //just in case it hasn't been cleared up last time map.clear(); @@ -122,7 +113,7 @@ public class FilePerKeyMapTest { @Test public void perfTest(){ //There are no entries in the map so null should be returned - FilePerKeyMap map = new FilePerKeyMap<>("/tmp/filepermaptests"); + FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests"); //just in case it hasn't been cleared up last time map.clear(); @@ -133,8 +124,8 @@ public class FilePerKeyMapTest { } String value = sb.toString(); - //warmup - for (int i = 0; i < 1000; i++) { + //small warm-up + for (int i = 0; i < 10; i++) { map.put("big file", value); } @@ -147,4 +138,30 @@ public class FilePerKeyMapTest { System.out.println("Time to update "+ iterations + " iterations " + (System.currentTimeMillis()-time)); } + @Test + public void eventTest(){ + //There are no entries in the map so null should be returned + FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests"); + + //just in case it hasn't been cleared up last time + map.clear(); + + map.registerForEvents(new FPMListener() { + @Override + public void onEvent(FPMEvent e) { + System.out.println(e); + } + + }); + + map.put("one", "one"); + map.put("one", "one"); + + try { + Thread.sleep(300000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + } From 5da2a603b5d089aa777bd73a42ed427dab76727e Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 24 Apr 2015 09:28:18 +0100 Subject: [PATCH 2/8] simplified --- .../java/net/openhft/chronicle/map/FilePerKeyMap.java | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java index b94d7ab71..32faf7719 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java @@ -280,12 +280,9 @@ public class FilePerKeyMap implements Map { } } - private boolean isProgrammaticUpdate(File file){ - if(lastModifiedByProgram.containsKey(file)){ - return file.lastModified() == lastModifiedByProgram.get(file) ? true : false; - }else{ - return false; - } + private boolean isProgrammaticUpdate(File file) { + return lastModifiedByProgram.containsKey(file) + && (file.lastModified() == lastModifiedByProgram.get(file)); } } From 53baa8b09b91c1967f850d45048f8d7fde345b66 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 24 Apr 2015 10:45:49 +0100 Subject: [PATCH 3/8] simplified test --- .../openhft/chronicle/map/TcpReplicator.java | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index d18fa86e5..b8e81bd93 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -897,16 +897,13 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo serverSocket.bind(details.address()); // these can be run on this thread - addPendingRegistration(new Runnable() { - @Override - public void run() { - final Attached attached = new Attached(); - attached.connector = ServerConnector.this; - try { - serverChannel.register(TcpReplicator.this.selector, OP_ACCEPT, attached); - } catch (ClosedChannelException e) { - LOG.debug("", e); - } + addPendingRegistration(() -> { + final Attached attached = new Attached(); + attached.connector = ServerConnector.this; + try { + serverChannel.register(TcpReplicator.this.selector, OP_ACCEPT, attached); + } catch (ClosedChannelException e) { + LOG.debug("", e); } }); From 2bd94c3b420503ff5e33a2bd209b7cec99af360a Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 24 Apr 2015 16:14:33 +0100 Subject: [PATCH 4/8] cleanups and added Map support to wire --- .../map/ClientWiredStatelessChronicleMap.java | 19 +++++++------------ .../net/openhft/chronicle/map/EngineMap.java | 6 +++--- .../chronicle/map/MapStatelessClient.java | 2 +- 3 files changed, 11 insertions(+), 16 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index f5c6ecf05..664d34015 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -40,6 +40,7 @@ import static java.util.Collections.emptyList; import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields.reply; import static net.openhft.chronicle.map.MapWireHandler.EventId; import static net.openhft.chronicle.map.MapWireHandler.EventId.*; +import static net.openhft.chronicle.map.VanillaChronicleMap.newInstance; /** @@ -51,24 +52,19 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient VOID_PARAMETERS = out -> out.marshallable(AbstactStatelessClient.EMPTY); private final Class vClass; - protected Class kClass; - private boolean putReturnsNull; private boolean removeReturnsNull; - // used with toString() - private static final int MAX_NUM_ENTRIES = 20; public ClientWiredStatelessChronicleMap( @NotNull final ClientWiredChronicleMapStatelessBuilder config, @NotNull final Class kClass, @NotNull final Class vClass, @NotNull final String channelName, - ClientWiredStatelessTcpConnectionHub hub) { + @NotNull final ClientWiredStatelessTcpConnectionHub hub) { super(channelName, hub, "MAP", 0, vClass); this.putReturnsNull = config.putReturnsNull(); @@ -96,12 +92,12 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient extends MapStatelessClient extends MapStatelessClient writeField(out, (K) key)); + return proxyReturnBoolean(containsKey, out -> out.object(out)); } @NotNull @@ -233,7 +228,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient writeField(out, (V) value)); + return proxyReturnBoolean(containsValue, out -> out.object(out)); } public void putAll(@NotNull Map map) { diff --git a/src/main/java/net/openhft/chronicle/map/EngineMap.java b/src/main/java/net/openhft/chronicle/map/EngineMap.java index 72d769a1f..11f4013e9 100644 --- a/src/main/java/net/openhft/chronicle/map/EngineMap.java +++ b/src/main/java/net/openhft/chronicle/map/EngineMap.java @@ -3,6 +3,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.hash.function.SerializableFunction; import net.openhft.chronicle.wire.TextWire; +import net.openhft.chronicle.wire.ValueOut; import net.openhft.chronicle.wire.Wire; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; @@ -80,7 +81,7 @@ public class EngineMap implements ChronicleMap { if (b == null) return null; final Wire wire = toWire(); - AbstactStatelessClient.writeField(b, wire.getValueOut()); + ((ValueOut) b).object(b); wire.bytes().flip(); return toWire().getValueIn().bytes(); @@ -160,13 +161,12 @@ public class EngineMap implements ChronicleMap { return null; final Wire wire = toWire(bytes); buffer.flip(); - return AbstactStatelessClient.readObject(wire.getValueIn(), null, eClass); + return wire.getValueIn().object(eClass); } @Override public V put(final K key, final V value) { nullCheck(key); - // nullCheck(value); return toObject(vClass, () -> map.put(bytes(key), bytes(value))); } diff --git a/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java b/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java index e9e0153db..582ab89c4 100644 --- a/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java +++ b/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java @@ -57,7 +57,7 @@ public abstract class MapStatelessClient @NotNull final E eventId, Object field) { final long startTime = System.currentTimeMillis(); - final long tid = sendEvent(startTime, eventId, out -> writeField(out, field)); + final long tid = sendEvent(startTime, eventId, out -> out.object(out)); if (eventReturnsNull(eventId)) return null; From 1aa888f5f5aca0e7d5215a8e6daadd9d473dd17c Mon Sep 17 00:00:00 2001 From: danielshaya Date: Fri, 24 Apr 2015 16:53:08 +0100 Subject: [PATCH 5/8] Initial commit for FilePerMap implementation. --- .../openhft/chronicle/map/FPMListener.java | 10 ---- .../chronicle/map/FilePerKeyChronicleMap.java | 46 +++++++++---------- .../openhft/chronicle/map/FilePerKeyMap.java | 24 ++++++---- .../chronicle/map/FilePerKeyMapTest.java | 23 ++++++---- 4 files changed, 51 insertions(+), 52 deletions(-) delete mode 100644 src/main/java/net/openhft/chronicle/map/FPMListener.java diff --git a/src/main/java/net/openhft/chronicle/map/FPMListener.java b/src/main/java/net/openhft/chronicle/map/FPMListener.java deleted file mode 100644 index 7fba5460b..000000000 --- a/src/main/java/net/openhft/chronicle/map/FPMListener.java +++ /dev/null @@ -1,10 +0,0 @@ -package net.openhft.chronicle.map; - -/** - * Created by daniel on 23/04/15. - */ -public class FPMListener { - public void onEvent(FPMEvent event){ - - } -} diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java index 6648c4c90..84cf435e4 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java @@ -8,7 +8,6 @@ import java.io.IOException; import java.util.Collection; import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; import java.util.function.Predicate; @@ -16,16 +15,13 @@ import java.util.function.Predicate; * Created by daniel on 22/04/15. */ public class FilePerKeyChronicleMap implements ChronicleMap { - private final String dir; - private final Map lastModifiedByProgram = new ConcurrentHashMap<>(); private final ChronicleMap chMap; private final FilePerKeyMap fpMap; public FilePerKeyChronicleMap(String dir) { - this.dir = dir; - fpMap = new FilePerKeyMap(dir); - chMap = ChronicleMapBuilder.of(String.class, String.class) - .entries(500).averageValueSize(2_000_000).create(); + fpMap = new FilePerKeyMap(dir); + chMap = ChronicleMapBuilder.of(String.class, String.class) + .entries(500).averageValueSize(2_000_000).create(); } @Override @@ -55,17 +51,20 @@ public class FilePerKeyChronicleMap implements ChronicleMap { @Override public String put(String key, String value) { - return fpMap.put(key, value); + fpMap.put(key, value); + return chMap.put(key,value); } @Override public String remove(Object key) { + chMap.remove(key); return fpMap.remove(key); } @Override public void putAll(Map m) { fpMap.putAll(m); + chMap.putAll(m); } @Override @@ -93,27 +92,28 @@ public class FilePerKeyChronicleMap implements ChronicleMap { @Override public String getUsing(String key, String 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); } @NotNull @Override public MapKeyContext acquireContext(@NotNull String key, @NotNull String usingValue) { - throw new UnsupportedOperationException(); + return chMap.acquireContext(key,usingValue); } @Override - public R getMapped( String key, @NotNull SerializableFunction function) { - throw new UnsupportedOperationException(); + public R getMapped(String key, @NotNull SerializableFunction function) { + return chMap.getMapped(key, function); } @Override - public String putMapped( String key, @NotNull UnaryOperator unaryOperator) { + public String putMapped(String key, @NotNull UnaryOperator unaryOperator) { throw new UnsupportedOperationException(); } @@ -129,17 +129,17 @@ public class FilePerKeyChronicleMap implements ChronicleMap { @Override public String newValueInstance() { - return null; + return new String(); } @Override public String newKeyInstance() { - throw new UnsupportedOperationException(); + return new String(); } @Override public Class valueClass() { - throw new UnsupportedOperationException(); + return String.class; } @Override @@ -153,28 +153,28 @@ public class FilePerKeyChronicleMap implements ChronicleMap { } @Override - public MapKeyContext context( String key) { - throw new UnsupportedOperationException(); + public MapKeyContext context(String key) { + return chMap.context(key); } @Override public Class keyClass() { - throw new UnsupportedOperationException(); + return String.class; } @Override public boolean forEachEntryWhile(Predicate> predicate) { - throw new UnsupportedOperationException(); + return chMap.forEachEntryWhile(predicate); } @Override public void forEachEntry(Consumer> action) { - throw new UnsupportedOperationException(); + chMap.forEachEntry(action); } @Override public void close() { - throw new UnsupportedOperationException(); + chMap.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 b94d7ab71..cd0925644 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java @@ -8,6 +8,7 @@ import java.io.IOException; import java.nio.file.*; import java.util.*; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -18,7 +19,8 @@ public class FilePerKeyMap implements Map { private final String dir; private final Path dirPath; private final Map lastModifiedByProgram = new ConcurrentHashMap<>(); - private final List listeners = new ArrayList<>(); + private final List> listeners = new ArrayList<>(); + private final FPMWatcher fileFpmWatcher = new FPMWatcher(); public FilePerKeyMap(String dir){ this.dir = dir; @@ -28,16 +30,20 @@ public class FilePerKeyMap implements Map { }catch (IOException e){ throw new RuntimeException(e); } - new FPMWatcher().start(); + fileFpmWatcher.start(); } - public void registerForEvents(FPMListener listener){ + public void registerForEvents(Consumer listener){ listeners.add(listener); } + public void unregisterForEvents(Consumer listener){ + listeners.remove(listener); + } + private void fireEvent(FPMEvent event){ - for (FPMListener listener : listeners) { - listener.onEvent(event); + for (Consumer listener : listeners) { + listener.accept(event); } } @@ -162,6 +168,9 @@ public class FilePerKeyMap implements Map { } } + public void close(){ + fileFpmWatcher.interrupt(); + } private static class FPMEntry implements Entry { @@ -238,7 +247,6 @@ public class FilePerKeyMap implements Map { if (kind == StandardWatchEventKinds.OVERFLOW) { - System.out.println("OVERFLOW"); continue; } else if (kind == StandardWatchEventKinds.ENTRY_CREATE) { Path p = dirPath.resolve(fileName); @@ -266,12 +274,10 @@ public class FilePerKeyMap implements Map { } } } - } catch (InterruptedException e) { - System.out.println("Watcher service exiting"); return; }finally{ - key.reset(); + if(key!=null)key.reset(); } } } catch (IOException e) { diff --git a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java index fa3590ea3..56c75ae50 100644 --- a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java @@ -108,6 +108,10 @@ public class FilePerKeyMapTest { } catch (InterruptedException e) { e.printStackTrace(); } + + map.clear(); + map.close(); + } @Test @@ -136,32 +140,31 @@ public class FilePerKeyMapTest { map.put("big file", value); } System.out.println("Time to update "+ iterations + " iterations " + (System.currentTimeMillis()-time)); + + map.clear(); + map.close(); } @Test - public void eventTest(){ + public void eventTest() { //There are no entries in the map so null should be returned FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests"); //just in case it hasn't been cleared up last time map.clear(); - map.registerForEvents(new FPMListener() { - @Override - public void onEvent(FPMEvent e) { - System.out.println(e); - } - - }); + map.registerForEvents(e -> System.out.println(e)); map.put("one", "one"); map.put("one", "one"); try { - Thread.sleep(300000); + Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } - } + map.clear(); + map.close(); + } } From df97973983d83dd80d2e9e9851be40d0fe81d130 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 24 Apr 2015 18:50:06 +0100 Subject: [PATCH 6/8] fixed issue with EngineMap --- src/main/java/net/openhft/chronicle/map/EngineMap.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/EngineMap.java b/src/main/java/net/openhft/chronicle/map/EngineMap.java index 11f4013e9..0a25773df 100644 --- a/src/main/java/net/openhft/chronicle/map/EngineMap.java +++ b/src/main/java/net/openhft/chronicle/map/EngineMap.java @@ -3,7 +3,6 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.hash.function.SerializableFunction; import net.openhft.chronicle.wire.TextWire; -import net.openhft.chronicle.wire.ValueOut; import net.openhft.chronicle.wire.Wire; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; @@ -81,7 +80,7 @@ public class EngineMap implements ChronicleMap { if (b == null) return null; final Wire wire = toWire(); - ((ValueOut) b).object(b); + wire.getValueOut().object(b); wire.bytes().flip(); return toWire().getValueIn().bytes(); From 12c9f1d11f4d5609a4aa53b74faef7208970d1d2 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 24 Apr 2015 19:46:56 +0100 Subject: [PATCH 7/8] fixed issue with EngineMap --- .../java/net/openhft/chronicle/map/MapStatelessClient.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java b/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java index 582ab89c4..7f8b1591b 100644 --- a/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java +++ b/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java @@ -54,10 +54,11 @@ public abstract class MapStatelessClient @Nullable protected R proxyReturnObject(@NotNull final Class rClass, - @NotNull final E eventId, Object field) { + @NotNull final E eventId, + @NotNull final Object o) { final long startTime = System.currentTimeMillis(); - final long tid = sendEvent(startTime, eventId, out -> out.object(out)); + final long tid = sendEvent(startTime, eventId, out -> out.object(o)); if (eventReturnsNull(eventId)) return null; From bdc6d0df905fb04240c148b3e46ad47d79ee5405 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 24 Apr 2015 20:15:00 +0100 Subject: [PATCH 8/8] fixed issue with EngineMap --- .../chronicle/map/ClientWiredStatelessChronicleMap.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 664d34015..06aa7adce 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -219,7 +219,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient out.object(out)); + return proxyReturnBoolean(containsKey, out -> out.object(key)); } @NotNull @@ -228,7 +228,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient out.object(out)); + return proxyReturnBoolean(containsValue, out -> out.object(value)); } public void putAll(@NotNull Map map) {