diff --git a/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java b/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java index c924eca78..ad82e9677 100644 --- a/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java +++ b/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java @@ -26,7 +26,7 @@ import java.util.concurrent.locks.Lock; /** * Abstracts inter-process lock of {@link ChronicleHash} data: segments, entries. * - *

This lock is not reentrant, but kind of "saturating": multiple {@link #lock()} calls has + *

This lock is not reentrant, but kind of "saturating": multiple {@link #lock()} calls have * the same effect as a single call. Likewise {@link #unlock()} -- multiple unlocks, or "unlocking" * when the lock isn't actually held, has no negative effects. * diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 0015f8ede..3814c43b5 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -1360,7 +1360,9 @@ public final class ChronicleMapBuilder implements try { // pushingToMapEventListener(); VanillaChronicleMap map = newMap(singleHashReplication, channel); - OS.warnOnWindows(map.sizeInBytes()); + if(OS.warnOnWindows(map.sizeInBytes())){ + throw new IllegalStateException("Windows cannot support this configuration"); + } BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE, map.sizeInBytes(), true); map.createMappedStoreAndSegments(bytesStore); diff --git a/src/main/java/net/openhft/chronicle/map/FPMEvent.java b/src/main/java/net/openhft/chronicle/map/FPMEvent.java deleted file mode 100644 index c2153b050..000000000 --- a/src/main/java/net/openhft/chronicle/map/FPMEvent.java +++ /dev/null @@ -1,64 +0,0 @@ -package net.openhft.chronicle.map; - -/** - * Created by daniel on 23/04/15. - */ -public class FPMEvent { - private EventType eventType; - private String key; - private String value; - private String lastValue; - - public FPMEvent(EventType eventType, String key, String lastValue, String value) { - this.eventType = eventType; - this.key = key; - this.value = value; - this.lastValue = lastValue; - } - - public String getLastValue() { - return lastValue; - } - - public void setLastValue(String lastValue) { - this.lastValue = lastValue; - } - - public EventType getEventType() { - return eventType; - } - - public void setEventType(EventType eventType) { - this.eventType = eventType; - } - - 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 + - ", key='" + key + '\'' + - ", value='" + value + '\'' + - ", lastValue='" + lastValue + '\'' + - '}'; - } - - public enum EventType {NEW, UPDATE, DELETE} - - -} diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java deleted file mode 100644 index 2d4ae6007..000000000 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java +++ /dev/null @@ -1,227 +0,0 @@ -package net.openhft.chronicle.map; - -import net.openhft.chronicle.hash.Data; -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.function.Consumer; -import java.util.function.Predicate; - -/** - * Created by daniel on 22/04/15. - */ -public class FilePerKeyChronicleMap implements ChronicleMap { - // private final ChronicleMap chMap; - private final FilePerKeyMap fpMap; - - public FilePerKeyChronicleMap(String dir) throws IOException { - fpMap = new FilePerKeyMap(dir); - 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 - 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); -// 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 - 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(); -// 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(); -// return chMap.getMapped(key, function); - } - - @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 new String(); - } - - @Override - public String newKeyInstance() { - return new String(); - } - - @Override - public Class valueClass() { - return String.class; - } - - @Override - public File file() { - throw new UnsupportedOperationException(); - } - - @Override - public long longSize() { - return size(); - } - - @Override - public MapKeyContext context(String key) { - throw new UnsupportedOperationException(); - // return chMap.context(key); - } - - @Override - public Class keyClass() { - return String.class; - } - - @NotNull - @Override - public ExternalMapQueryContext queryContext(String key) { - throw new UnsupportedOperationException(); - } - - @NotNull - @Override - public ExternalMapQueryContext queryContext(Data key) { - throw new UnsupportedOperationException(); - } - - @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() { - fpMap.close(); - } - - @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 deleted file mode 100644 index 758c6b291..000000000 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java +++ /dev/null @@ -1,358 +0,0 @@ -package net.openhft.chronicle.map; - -import com.sun.nio.file.SensitivityWatchEventModifier; -import net.openhft.chronicle.core.util.ThrowingFunction; -import org.jetbrains.annotations.NotNull; - -import java.io.*; -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; - -/** - * A {@link Map} implementation that stores each entry as a file in a - * directory. The key is the file name and the value - * is the contents of the file. This map will only handle String's. - *

- * The class is effectively an abstraction over a directory in the file system. - * Therefore when the underlying files are changed an event will be fired to those - * registered for notifications. - * Since every write to this map will cause a change to underlying - * file system the event will distinguish between a programmatic event - * (i.e one caused my the actions of the map itself) and an event that - * has been triggered as a direct result of a file being manipulated outside - * this class. - *

- * Updates will be fired every time the file is saved but will be suppressed - * if the value has not changed. To avoid temporary files (e.g. if edited in vi) - * being included in the map, any file starting with a '.' will be ignored. - *

- * Note the {@link WatchService} is extremely OS dependant. Mas OSX registers - * 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, Closeable { - private final Path dirPath; - private final Map lastFile = new ConcurrentHashMap<>(); - - 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); - 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) { - listeners.add(listener); - } - - public void unregisterForEvents(Consumer listener) { - listeners.remove(listener); - } - - 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); - } - } - - @Override - public int size() { - return (int) getFiles().count(); - } - - @Override - public boolean isEmpty() { - return size() == 0; - } - - @Override - public boolean containsKey(Object key) { - return getFiles().anyMatch(p -> p.getFileName().toString().equals(key)); - } - - @Override - public boolean containsValue(Object value) { - return getFiles().anyMatch(p -> getFileContents(p).equals(value)); - } - - @Override - public String get(Object key) { - Path path = dirPath.resolve((String) key); - return getFileContents(path); - } - - @Override - public String put(String key, String value) { - if (closed) throw new IllegalStateException("closed"); - Path path = dirPath.resolve(key); - String existingValue = putReturnsNull ? null : getFileContents(path); - writeToFile(path, value); - - return existingValue; - } - - @Override - public String remove(Object key) { - if (closed) throw new IllegalStateException("closed"); - String existing = get(key); - if (existing != null) { - deleteFile(dirPath.resolve((String) key)); - } - return existing; - } - - @Override - public void putAll(Map m) { - m.entrySet().stream().forEach(e -> put(e.getKey(), e.getValue())); - } - - @Override - public void clear() { - getFiles().forEach(this::deleteFile); - } - - @NotNull - @Override - public Set keySet() { - return getFiles().map(p -> p.getFileName().toString()) - .collect(Collectors.toSet()); - } - - @NotNull - @Override - public Collection values() { - return getFiles().map(p -> getFileContents(p)) - .collect(Collectors.toSet()); - - } - - @NotNull - @Override - public Set> entrySet() { - return getFiles().map(p -> - (Entry) new FPMEntry(p.getFileName().toString(), getFileContents(p))) - .collect(Collectors.toSet()); - } - - - private Stream getFiles() { - try { - return Files.walk(dirPath).filter(p -> !Files.isDirectory(p)); - } 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); - } - } - - 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) { - File key = path.toFile(); - key.delete(); - } - - public void close() { - closed = true; - fileFpmWatcher.interrupt(); - } - - 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) { - this.key = key; - this.value = value; - } - - @Override - public String getKey() { - return key; - } - - @Override - public String getValue() { - return value; - } - - @Override - public String setValue(String value) { - String lastValue = this.value; - this.value = value; - return lastValue; - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - - FPMEntry fpmEntry = (FPMEntry) o; - - if (key != null ? !key.equals(fpmEntry.key) : fpmEntry.key != null) return false; - return !(value != null ? !value.equals(fpmEntry.value) : fpmEntry.value != null); - } - - @Override - public int hashCode() { - int result = key != null ? key.hashCode() : 0; - result = 31 * result + (value != null ? value.hashCode() : 0); - return result; - } - } - - private class FPMWatcher implements Runnable { - private final WatchService watcher; - - 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(".")) { - //this avoids temporary files being added to the map - continue; - } - - if (kind == StandardWatchEventKinds.ENTRY_CREATE) { - Path p = dirPath.resolve(fileName); - 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); - - 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) { - 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; - } - - fireEvent(new FPMEvent(FPMEvent.EventType.UPDATE, mapKey, lastVal, mapVal)); - } catch (FileNotFoundException ignored) { - } - } - } - } catch (InterruptedException e) { - return; - } finally { - if (key != null) key.reset(); - } - } - } catch (Throwable e) { - if (!closed) - e.printStackTrace(); - } - } - } -} - -class FileRecord { - final long timestamp; - final String contents; - - FileRecord(long timestamp, String contents) { - this.timestamp = timestamp; - this.contents = contents; - } -} diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 3defb8fce..597c5c7b6 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -442,7 +442,7 @@ public class ReplicatedChronicleMap remoteOpContext = mapContext()) { remoteOpContext.initReplicationInput(source); - remoteOpContext.processReplicatedEvent(); + //remoteOpContext.processReplicatedEvent(); } } diff --git a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java deleted file mode 100644 index 262b357cb..000000000 --- a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java +++ /dev/null @@ -1,166 +0,0 @@ -package net.openhft.chronicle.map; - -import org.junit.BeforeClass; -import org.junit.Ignore; -import org.junit.Test; - -import java.io.IOException; -import java.util.*; -import java.util.function.Consumer; - -import static org.junit.Assert.*; - -/** - * JUnit test class to support {@link FilePerKeyMapTest} - */ -public class FilePerKeyMapTest { - private static FilePerKeyMap map; - @BeforeClass - public static void createMap() throws IOException { - map = new FilePerKeyMap("/tmp/filepermaptests"); - //just in case it hasn't been cleared up last time - map.clear(); - } - /** - * Testing all the methods of the map with simple tests. - */ - @Test - @Ignore - public void testMapMethods() { - String val = map.put("one", "test1"); - assertEquals(val, null); - - //After the entry has been written the value returned should be - //the previous value - val = map.put("one", "test2"); - assertEquals(val, "test1"); - - //Check that get returns the latest value - val = map.get("one"); - assertEquals("test2", val); - - assertTrue(map.containsKey("one")); - assertFalse(map.containsKey("two")); - assertTrue(map.containsValue("test2")); - assertFalse(map.containsValue("test5")); - - //Check the size is one - int size = map.size(); - assertEquals(1, size); - assertFalse(map.isEmpty()); - - map.clear(); - //Check the size is now 0 - size = map.size(); - assertEquals(0, size); - - assertTrue(map.isEmpty()); - - map.put("one", "test1"); - map.put("two", "test2"); - map.put("three", "test3"); - assertEquals(3, map.size()); - - val = map.remove("two"); - assertEquals("test2", val); - - assertEquals(2, map.size()); - - val = map.remove("four");//doesn't exist - assertEquals(null, val); - - map.clear(); - - Map copyFrom = new HashMap<>(); - copyFrom.put("five", "test5"); - copyFrom.put("six", "test6"); - copyFrom.put("seven", "test7"); - map.putAll(copyFrom); - assertEquals(3, map.size()); - - Set set = map.keySet(); - assertEquals(3, set.size()); - assertTrue(set.contains("five")); - assertTrue(set.contains("six")); - assertTrue(set.contains("seven")); - - set = (Set) map.values(); - assertEquals(3, set.size()); - assertTrue(set.contains("test5")); - assertTrue(set.contains("test6")); - assertTrue(set.contains("test7")); - - Set entryset = (Set) map.entrySet(); - assertEquals(3, entryset.size()); - for (Iterator it = entryset.iterator(); it.hasNext(); ) { - Map.Entry entry = it.next(); - if (entry.getKey().equals("five")) { - assertEquals(entry.getValue(), "test5"); - } else if (entry.getKey().equals("six")) { - assertEquals(entry.getValue(), "test6"); - } else if (entry.getKey().equals("seven")) { - assertEquals(entry.getValue(), "test7"); - } else { - //should never get here!! - assertTrue(false); - } - } - - map.clear(); - } - - @Test - public void perfTest(){ - StringBuilder sb = new StringBuilder(); - for (int i = 0; i < 2_000_000; i++) { - sb.append('x'); - } - String value = sb.toString(); - - //small warm-up - for (int i = 0; i < 10; i++) { - map.put("big file", value); - } - - - long time = System.currentTimeMillis(); - int iterations = 50; - for (int i = 0; i < iterations; i++) { - map.put("big file", value); - } - System.out.println("Time to update " + iterations + " iterations " + (System.currentTimeMillis() - time)); - - map.clear(); - } - - @Ignore //This doesn't work on all OS - @Test - public void eventTest() throws InterruptedException { - final List events = new ArrayList<>(); - - Consumer fpmEventConsumer = (FPMEvent e) -> { - System.out.println(e); - events.add(e); - }; - map.registerForEvents(fpmEventConsumer); - - map.put("one", "one"); - map.put("one", "two"); - map.remove("one"); - map.put("one", "one"); - - //Allow all events to be played through - Thread.sleep(3000); - //Check the events - assertEquals(4, events.size()); - - assertEquals(FPMEvent.EventType.NEW,events.get(0).getEventType()); - assertEquals(FPMEvent.EventType.UPDATE, events.get(1).getEventType()); - assertEquals(FPMEvent.EventType.DELETE, events.get(2).getEventType()); - assertEquals(FPMEvent.EventType.NEW, events.get(3).getEventType()); - - - map.unregisterForEvents(fpmEventConsumer); - map.clear(); - } -}