Add file locking, work on #229

src/main/java/net/openhft/chronicle/hash/ChronicleFileLockException.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2012-2018 Chronicle Map Contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash;
+
+/**
+ * This exception is thrown, when a Chronicle Hash cannot acquire a required file lock.
+ *
+ */
+public final class ChronicleFileLockException extends RuntimeException {
+    private static final long serialVersionUID = -2034623786298623984L;
+
+    /**
+     * Constructs a new {@code ChronicleHashLockException} with the specified cause.
+     *
+     * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method)
+     */
+    public ChronicleFileLockException(Throwable cause) {
+        super(cause);
+    }
+
+    /**
+     * Constructs a new {@code ChronicleHashLockException} with the specified detail
+     * message.
+     *
+     * @param message the detail message. The detail message is saved for later retrieval by the
+     *                {@link #getMessage()} method.
+     */
+    public ChronicleFileLockException(String message) {
+        super(message);
+    }
+
+    public ChronicleFileLockException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java

@@ -51,6 +51,7 @@ import java.lang.ref.WeakReference;
 import java.lang.reflect.Type;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -66,6 +67,7 @@ import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.*;
 import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.format;
 import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.report;
 
+
 public abstract class VanillaChronicleHash<K,
                                            C extends HashEntry<K>, SC extends HashSegmentContext<K, ?>,
                                            ECQ extends ExternalHashQueryContext<K>>
@@ -153,6 +155,12 @@ public abstract class VanillaChronicleHash<K,
     private transient Cleaner cleaner;
     private transient VanillaGlobalMutableState globalMutableState;
 
+    /**
+     * The fileLock is used to prevent recover actions
+     * from accessing the mapped file concurrently.
+     */
+    @Nullable private transient FileLock fileLock;
+
     public VanillaChronicleHash(@NotNull final ChronicleMapBuilder<K, ?> builder) {
         // Version
         dataFileVersion = BuildVersion.version();
@@ -401,7 +409,6 @@ public abstract class VanillaChronicleHash<K,
         identity = new Identity();
     }
 
-    // 999
     public final void initBeforeMapping(@NotNull final File file,
                                         @NotNull final RandomAccessFile raf,
                                         final long headerEnd,
@@ -412,9 +419,8 @@ public abstract class VanillaChronicleHash<K,
         if (!createdOrInMemory) {
             // This block is for reading segmentHeadersOffset before main mapping
             // After the mapping globalMutableState value's bytes are reassigned
-            ByteBuffer globalMutableStateBuffer =
-                    ByteBuffer.allocate((int) globalMutableState.maxSize());
-            FileChannel fileChannel = raf.getChannel();
+            final ByteBuffer globalMutableStateBuffer = ByteBuffer.allocate((int) globalMutableState.maxSize());
+            final FileChannel fileChannel = raf.getChannel();
             while (globalMutableStateBuffer.remaining() > 0) {
                 if (fileChannel.read(globalMutableStateBuffer,
                         this.headerSize + GLOBAL_MUTABLE_STATE_VALUE_OFFSET +
@@ -424,9 +430,9 @@ public abstract class VanillaChronicleHash<K,
             }
             globalMutableStateBuffer.flip();
             //noinspection unchecked
-            globalMutableState.bytesStore(BytesStore.wrap(globalMutableStateBuffer), 0,
-                    globalMutableState.maxSize());
+            globalMutableState.bytesStore(BytesStore.wrap(globalMutableStateBuffer), 0, globalMutableState.maxSize());
         }
+
     }
 
     public final void createInMemoryStoreAndSegments(@NotNull final ChronicleHashResources resources) {
@@ -1109,4 +1115,53 @@ public abstract class VanillaChronicleHash<K,
             return VanillaChronicleHash.this;
         }
     }
+
+    /*
+    protected void acquireSharedFileLock() {
+        assertFileLockEmpty();
+        System.out.println("SHARED "+file());
+
+        try {
+            fileLock = raf.getChannel().tryLock(0, Long.MAX_VALUE, true);
+            System.out.println("fileLock.isShared = " + fileLock.isShared());
+        } catch (Exception e) {
+            throw new ChronicleFileLockException("Unable to acquire a shared file lock for " + file() + ". " +
+                    "Make sure another process is not recovering the map.", e);
+        }
+    }
+
+    protected void acquireExclusiveFileLock() {
+        assertFileLockEmpty();
+        System.out.println("EXCLUSIVE "+file());
+        try {
+            fileLock = raf.getChannel().tryLock(0, Long.MAX_VALUE, false);
+            System.out.println("fileLock.isShared = " + fileLock.isShared());
+        } catch (Exception e) {
+            throw new ChronicleFileLockException("Unable to acquire an exclusive file lock for " + file() + ". " +
+                    "Make sure no other process is using the map.", e);
+        }
+    }
+
+    protected void releaseFileLock() {
+        System.out.println("RELEASE "+file());
+        try {
+            requireFileLockPresent().release();
+        } catch (IOException e) {
+            throw new ChronicleFileLockException("Unable to release the file lock for " + file(), e);
+        }
+    }
+
+    private void assertFileLockEmpty() {
+        if (fileLock != null)
+            throw new ChronicleFileLockException("A file lock instance already exists for the file " + file());
+    }
+
+    private FileLock requireFileLockPresent() {
+        if (fileLock == null)
+            throw new ChronicleFileLockException("No file lock instance exists for the file " + file());
+        return fileLock;
+    }
+
+     */
+
 }

src/main/java/net/openhft/chronicle/hash/impl/util/CanonicalRandomAccessFiles.java

@@ -62,6 +62,8 @@ public final class CanonicalRandomAccessFiles {
         });
     }
 
+    // This class is not thread-safe but instances
+    // are protected by means of the CANONICAL_RAFS map
     private static final class RafReference {
         private final RandomAccessFile raf;
         private int refCount;

src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java

@@ -1773,13 +1773,15 @@ public final class ChronicleMapBuilder<K, V> implements
     }
 
     private VanillaChronicleMap<K, V, ?> createWithNewFile(@NotNull final VanillaChronicleMap<K, V, ?> map,
-                                                           @NotNull final File file,
+                                                           @NotNull final File canonicalFile,
                                                            @NotNull final RandomAccessFile raf,
                                                            @NotNull final ChronicleHashResources resources,
                                                            @NotNull final ByteBuffer headerBuffer,
                                                            final int headerSize) throws IOException {
-        map.initBeforeMapping(file, raf, headerBuffer.limit(), false);
+        map.initBeforeMapping(canonicalFile, raf, headerBuffer.limit(), false);
         map.createMappedStoreAndSegments(resources);
+        FileLockUtil.acquireSharedFileLock(canonicalFile, raf.getChannel());
+        map.addCloseable(() -> FileLockUtil.releaseFileLock(canonicalFile));
         commitChronicleMapReady(map, raf, headerBuffer, headerSize);
         return map;
     }
@@ -1817,6 +1819,7 @@ public final class ChronicleMapBuilder<K, V> implements
             final Wire wire = new TextWire(headerBytes);
             final VanillaChronicleMap<K, V, ?> map = wire.getValueIn().typedMarshallable();
             map.initBeforeMapping(file, raf, headerBuffer.limit(), recover);
+
             final long dataStoreSize = map.globalMutableState().getDataStoreSize();
             if (!recover && dataStoreSize > file.length()) {
                 throw new IOException("The file " + file + " the map is serialized from " +
@@ -1826,12 +1829,22 @@ public final class ChronicleMapBuilder<K, V> implements
             map.initTransientsFromBuilder(this);
             if (!recover) {
                 map.createMappedStoreAndSegments(resources);
+                FileLockUtil.acquireSharedFileLock(file, raf.getChannel());
             } else {
                 if (!headerWritten)
                     writeNotComplete(fileChannel, headerBuffer, headerSize);
-                map.recover(resources, corruptionListener, corruption);
+                try {
+                    FileLockUtil.acquireExclusiveFileLock(file, raf.getChannel());
+                    map.recover(resources, corruptionListener, corruption);
+                } finally {
+                    FileLockUtil.releaseFileLock(file);
+                }
+                // We are ready with exclusive access.
+                // Demote the lock from exclusive to shared
+                FileLockUtil.acquireSharedFileLock(file, fileChannel);
                 commitChronicleMapReady(map, raf, headerBuffer, headerSize);
             }
+            map.addCloseable(() -> FileLockUtil.releaseFileLock(file));
             return map;
         } catch (Throwable t) {
             if (recover && !(t instanceof IOException) &&
@@ -1894,7 +1907,7 @@ public final class ChronicleMapBuilder<K, V> implements
 
     private <E> double preMapConstruction(@NotNull final SerializationBuilder<E> builder,
                                           final double configuredAverageSize,
-                                          @NotNull final E average,
+                                          @Nullable final E average,
                                           @Nullable final E sample,
                                           @NotNull final String dim) {
         if (sample != null) {

src/main/java/net/openhft/chronicle/map/FileLockUtil.java

@@ -0,0 +1,158 @@
+package net.openhft.chronicle.map;
+
+import net.openhft.chronicle.hash.ChronicleFileLockException;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.util.concurrent.ConcurrentHashMap;
+
+public final class FileLockUtil {
+
+    private static final ConcurrentHashMap<File, FileLockReference> FILE_LOCKS = new ConcurrentHashMap<>();
+
+    private FileLockUtil() { }
+
+    public static void acquireSharedFileLock(@NotNull final File canonicalFile, @NotNull final FileChannel channel) {
+        FILE_LOCKS.compute(canonicalFile, (f, flr) ->
+                {
+                    try {
+                        if (flr == null)
+                            return new FileLockReference(channel.lock(0, Long.MAX_VALUE, true));
+                        else {
+                            if (!flr.fileLock.isShared()) {
+                                throw newUnableToAcquireSharedFileLockException(canonicalFile, null);
+                            }
+                            flr.reserve();
+                            return flr; // keep the old one
+                        }
+                    } catch (IOException e) {
+                        throw newUnableToAcquireSharedFileLockException(canonicalFile, e);
+                    }
+                }
+        );
+    }
+
+    public static void acquireExclusiveFileLock(@NotNull final File canonicalFile, @NotNull final FileChannel channel) {
+        FILE_LOCKS.compute(canonicalFile, (f, flr) ->
+                {
+                    if (flr == null) {
+                        try {
+                            final FileLock fileLock = channel.lock(0, Long.MAX_VALUE, false);
+                            return new FileLockReference(fileLock);
+                        } catch (IOException e) {
+                            throw newUnableToAcquireExclusiveFileLockException(canonicalFile, e);
+                        }
+                    } else {
+                        throw newUnableToAcquireExclusiveFileLockException(canonicalFile, null);
+                    }
+
+                }
+        );
+    }
+
+    public static void releaseFileLock(@NotNull final File canonicalFile) {
+        FILE_LOCKS.compute(canonicalFile, (f, flr) ->
+                {
+                    if (flr == null)
+                        throw new ChronicleFileLockException("Trying to release lock on file " + canonicalFile + " that did not exist");
+                    else {
+                        final int cnt = flr.release();
+                        if (cnt == 0)
+                            return null; // Remove the old one
+                        else
+                            return flr;
+                    }
+                }
+
+        );
+    }
+
+    public static void runExclusively(@NotNull final File canonicalFile,
+                                      @NotNull final FileChannel fileChannel,
+                                      @NotNull final Runnable fileIOAction) {
+
+        // Atomically acquire a FileLockReference
+        final FileLockReference fileLockRef = FILE_LOCKS.compute(canonicalFile, (f, flr) ->
+                {
+                    if (flr != null)
+                        throw new ChronicleFileLockException("A file lock instance already exists for the file " + canonicalFile);
+                    try {
+                        return new FileLockReference(fileChannel.lock());
+                    } catch (IOException e) {
+                        throw new ChronicleFileLockException(e);
+                    }
+                }
+        );
+
+        try {
+            fileIOAction.run();
+        } finally {
+            try {
+                fileLockRef.release();
+            } catch (Exception ignore) {
+            } finally {
+                FILE_LOCKS.remove(canonicalFile);
+            }
+        }
+    }
+
+    // This class is not thread-safe but instances
+    // are protected by means of the FILE_LOCKS map
+    private static final class FileLockReference {
+        private final FileLock fileLock;
+        private int refCount;
+
+        FileLockReference(@NotNull final FileLock fileLock) {
+            this.fileLock = fileLock;
+            refCount = 1;
+        }
+
+        int reserve() {
+            if (refCount == 0)
+                throw new IllegalStateException("Ref counter previously released");
+            return ++refCount;
+        }
+
+        int release() {
+            final int cnt = --refCount;
+            if (cnt == 0) {
+                try {
+                    fileLock.release();
+                } catch (IOException e) {
+                    throw new ChronicleFileLockException(e);
+                }
+            }
+            if (cnt < 0)
+                throw new IllegalStateException("Ref counter was " + cnt);
+
+            return cnt;
+        }
+
+        @Override
+        public String toString() {
+            return "FileLockReference{" +
+                    "fileLock=" + fileLock +
+                    ", refCount=" + refCount +
+                    '}';
+        }
+    }
+
+    static void dump() {
+        System.out.println(FILE_LOCKS);
+    }
+
+    private static ChronicleFileLockException newUnableToAcquireSharedFileLockException(@NotNull final File canonicalFile, @Nullable final Exception e) {
+        return new ChronicleFileLockException("Unable to acquire a shared file lock for " + canonicalFile + ". " +
+                "Make sure another process is not recovering the map.", e);
+    }
+
+    private static ChronicleFileLockException newUnableToAcquireExclusiveFileLockException(@NotNull final File canonicalFile, @Nullable final Exception e) {
+        return new ChronicleFileLockException("Unable to acquire an exclusive file lock for " + canonicalFile + ". " +
+                "Make sure no other process is using the map.", e);
+    }
+
+}