diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleFileLockException.java b/src/main/java/net/openhft/chronicle/hash/ChronicleFileLockException.java new file mode 100644 index 000000000..cd560c71f --- /dev/null +++ b/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); + } +} \ No newline at end of file diff --git a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java index ff5a67f15..dc52c45c9 100755 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/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, SC extends HashSegmentContext, ECQ extends ExternalHashQueryContext> @@ -153,6 +155,12 @@ public abstract class VanillaChronicleHash builder) { // Version dataFileVersion = BuildVersion.version(); @@ -401,7 +409,6 @@ public abstract class VanillaChronicleHash 0) { if (fileChannel.read(globalMutableStateBuffer, this.headerSize + GLOBAL_MUTABLE_STATE_VALUE_OFFSET + @@ -424,9 +430,9 @@ public abstract class VanillaChronicleHash implements } private VanillaChronicleMap createWithNewFile(@NotNull final VanillaChronicleMap 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 implements final Wire wire = new TextWire(headerBytes); final VanillaChronicleMap 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 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 implements private double preMapConstruction(@NotNull final SerializationBuilder builder, final double configuredAverageSize, - @NotNull final E average, + @Nullable final E average, @Nullable final E sample, @NotNull final String dim) { if (sample != null) { diff --git a/src/main/java/net/openhft/chronicle/map/FileLockUtil.java b/src/main/java/net/openhft/chronicle/map/FileLockUtil.java new file mode 100644 index 000000000..44834ad64 --- /dev/null +++ b/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_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); + } + +} \ No newline at end of file diff --git a/src/test/java/net/openhft/chronicle/map/FileLockUtilTest.java b/src/test/java/net/openhft/chronicle/map/FileLockUtilTest.java new file mode 100644 index 000000000..04d7f7eb4 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/FileLockUtilTest.java @@ -0,0 +1,108 @@ +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.ChronicleFileLockException; +import net.openhft.chronicle.hash.impl.util.CanonicalRandomAccessFiles; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.channels.FileChannel; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +public class FileLockUtilTest { + + private File canonicalFile; + private RandomAccessFile raf; + private FileChannel fileChannel; + + @Before + public void setUp() throws IOException { + canonicalFile = new File("file.lock").getCanonicalFile(); + canonicalFile.delete(); + canonicalFile.createNewFile(); + raf = CanonicalRandomAccessFiles.acquire(canonicalFile); + fileChannel = raf.getChannel(); + } + + @After + public void cleanup() throws IOException { + fileChannel.close(); + CanonicalRandomAccessFiles.release(canonicalFile); + } + + @Test + public void testShared() { + FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); + FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); + FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseFileLock(canonicalFile); + } + + @Test + public void testExclusiveNormalCase() { + FileLockUtil.acquireExclusiveFileLock(canonicalFile, fileChannel); + FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.acquireExclusiveFileLock(canonicalFile, fileChannel); + FileLockUtil.releaseFileLock(canonicalFile); + } + + @Test + public void testTryExclusiveButWasShared() { + FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); + try { + FileLockUtil.acquireExclusiveFileLock(canonicalFile, fileChannel); + fail(); + } catch (ChronicleFileLockException ignore) { + } + FileLockUtil.releaseFileLock(canonicalFile); + } + + @Test + public void testTrySharedButWasExclusive() { + FileLockUtil.acquireExclusiveFileLock(canonicalFile, fileChannel); + try { + FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); + fail(); + } catch (ChronicleFileLockException ignore) { + } + FileLockUtil.releaseFileLock(canonicalFile); + } + + @Test + public void testComplicated() { + FileLockUtil.acquireExclusiveFileLock(canonicalFile, fileChannel); + FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); + FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); + FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.acquireExclusiveFileLock(canonicalFile, fileChannel); + FileLockUtil.releaseFileLock(canonicalFile); + } + + @Test + public void testRunExclusively() { + FileLockUtil.dump(); + final AtomicInteger cnt = new AtomicInteger(); + FileLockUtil.runExclusively(canonicalFile, fileChannel, cnt::incrementAndGet); + assertEquals(1, cnt.get()); + } + + @Test + public void testRunExclusivelyButUsed() { + FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); + try { + FileLockUtil.runExclusively(canonicalFile, fileChannel, () -> {}); + fail(); + } catch (ChronicleFileLockException e) { + FileLockUtil.releaseFileLock(canonicalFile); + } + } + +} \ No newline at end of file diff --git a/src/test/java/net/openhft/chronicle/map/Issue229Test.java b/src/test/java/net/openhft/chronicle/map/Issue229Test.java new file mode 100644 index 000000000..5c7cdcfea --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/Issue229Test.java @@ -0,0 +1,60 @@ +/* + * 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.map; + +import net.openhft.chronicle.hash.ChronicleHashRecoveryFailedException; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; + +public class Issue229Test { + + private File mapFile; + + @Before + public void setup() { + mapFile = new File("test_map"); + } + + @After + public void cleanup() { + mapFile.delete(); + } + + @Test(expected = ChronicleHashRecoveryFailedException.class) + public void assureExclusiveAccess() throws IOException { + + try (ChronicleMap readMap = ChronicleMap + .of(Long.class, Long.class) + .entries(10) + .createPersistedTo(mapFile)) { + + + // It shall not be possible to recover since the + // file is open by the readMap + try (ChronicleMap recoverMap = ChronicleMap + .of(Long.class, Long.class) + .entries(10) + .recoverPersistedTo(mapFile, true)) { + + } + } + } +} \ No newline at end of file