diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 5cd4d83d9..ac78fd126 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -39,6 +39,8 @@ import net.openhft.chronicle.hash.serialization.impl.TypedMarshallableReaderWrit import net.openhft.chronicle.map.internal.AnalyticsHolder; import net.openhft.chronicle.map.replication.MapRemoteOperations; import net.openhft.chronicle.set.ChronicleSetBuilder; +import net.openhft.chronicle.threads.Pauser; +import net.openhft.chronicle.threads.TimingPauser; import net.openhft.chronicle.values.ValueModel; import net.openhft.chronicle.values.Values; import net.openhft.chronicle.wire.Marshallable; @@ -58,6 +60,7 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -1688,22 +1691,37 @@ public final class ChronicleMapBuilder implements if (raf.length() > 0) { result = openWithExistingFile(canonicalFile, raf, resources, recover, overrideBuilderConfig, corruptionListener); } else { - // Atomic* allows lambda modification final AtomicReference> map = new AtomicReference<>(); final AtomicReference headerBuffer = new AtomicReference<>(); - final AtomicBoolean newFile = new AtomicBoolean(); + final AtomicBoolean newFile = new AtomicBoolean(false); final FileChannel fileChannel = raf.getChannel(); - FileLockUtil.runExclusively(canonicalFile, fileChannel, () -> { - if (raf.length() == 0) { - map.set(newMap()); - headerBuffer.set(writeHeader(fileChannel, map.get())); - newFile.set(true); - } else { - newFile.set(false); + TimingPauser pauser = Pauser.balanced(); + + while (raf.length() == 0) { + final boolean locked = FileLockUtil.tryRunExclusively(canonicalFile, fileChannel, () -> { + // Double-checked locking + if (raf.length() == 0) { + map.set(newMap()); + headerBuffer.set(writeHeader(fileChannel, map.get())); + newFile.set(true); + } + }); + + if (locked) + break; + else { + try { + pauser.pause(10, TimeUnit.SECONDS); + } + catch (TimeoutException e) { + LOG.warn("Failed to write header: can't acquire exclusive file lock on empty file [" + canonicalFile + "] for 10 seconds", e); + + Jvm.rethrow(e); + } } - }); + } if (newFile.get()) { final int headerSize = headerBuffer.get().remaining(); @@ -1811,7 +1829,7 @@ public final class ChronicleMapBuilder implements map.initBeforeMapping(canonicalFile, raf, headerBuffer.limit(), false); map.createMappedStoreAndSegments(resources); FileLockUtil.acquireSharedFileLock(canonicalFile, raf.getChannel()); - map.addCloseable(() -> FileLockUtil.releaseFileLock(canonicalFile)); + map.addCloseable(() -> FileLockUtil.releaseSharedFileLock(canonicalFile)); commitChronicleMapReady(map, raf, headerBuffer, headerSize); return map; } @@ -1879,14 +1897,14 @@ public final class ChronicleMapBuilder implements FileLockUtil.acquireExclusiveFileLock(file, raf.getChannel()); map.recover(resources, corruptionListener, corruption); } finally { - FileLockUtil.releaseFileLock(file); + FileLockUtil.releaseExclusiveFileLock(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)); + map.addCloseable(() -> FileLockUtil.releaseSharedFileLock(file)); if (MAP_CREATION_DEBUG) { Jvm.warn().on(getClass(), " Created map [name=" + map.name() + ", size=" + map.longSize() + "] from file [canonizedMapDataFile=" + diff --git a/src/main/java/net/openhft/chronicle/map/FileLockUtil.java b/src/main/java/net/openhft/chronicle/map/FileLockUtil.java index 161877f77..dfc834a94 100644 --- a/src/main/java/net/openhft/chronicle/map/FileLockUtil.java +++ b/src/main/java/net/openhft/chronicle/map/FileLockUtil.java @@ -10,6 +10,7 @@ import java.io.File; import java.io.IOException; import java.nio.channels.FileChannel; import java.nio.channels.FileLock; +import java.nio.channels.OverlappingFileLockException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; @@ -18,15 +19,18 @@ public final class FileLockUtil { /** * Java file locks are maintained on a per JVM basis. So we need to manage them. */ + private static final String DISABLE_LOCKING = "chronicle.map.disable.locking"; private static final ConcurrentHashMap FILE_LOCKS = new ConcurrentHashMap<>(); - private static final boolean USE_LOCKING = !OS.isWindows() && !Jvm.getBoolean("chronicle.map.disable.locking"); + private static final boolean USE_EXCLUSIVE_LOCKING = !OS.isWindows() && !Jvm.getBoolean(DISABLE_LOCKING); + private static final boolean USE_SHARED_LOCKING = !OS.isWindows() && !Jvm.getBoolean(DISABLE_LOCKING) && + !"shared".equalsIgnoreCase(System.getProperty(DISABLE_LOCKING)); private static final AtomicBoolean LOCK_WARNING_PRINTED = new AtomicBoolean(); private FileLockUtil() { } public static void acquireSharedFileLock(@NotNull final File canonicalFile, @NotNull final FileChannel channel) { - if (USE_LOCKING) + if (USE_SHARED_LOCKING) FILE_LOCKS.compute(canonicalFile, (f, flr) -> { try { @@ -50,7 +54,7 @@ public final class FileLockUtil { } public static void acquireExclusiveFileLock(@NotNull final File canonicalFile, @NotNull final FileChannel channel) { - if (USE_LOCKING) + if (USE_EXCLUSIVE_LOCKING) FILE_LOCKS.compute(canonicalFile, (f, flr) -> { if (flr == null) { @@ -70,26 +74,90 @@ public final class FileLockUtil { printWarningTheFirstTime(); } - public static void releaseFileLock(@NotNull final File canonicalFile) { - if (USE_LOCKING) - 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 releaseSharedFileLock(@NotNull final File canonicalFile) { + if (USE_SHARED_LOCKING) + releaseFileLock0(canonicalFile); else printWarningTheFirstTime(); } + public static void releaseExclusiveFileLock(@NotNull final File canonicalFile) { + if (USE_EXCLUSIVE_LOCKING) + releaseFileLock0(canonicalFile); + else + printWarningTheFirstTime(); + } + + /** + * @deprecated Use {@link #releaseExclusiveFileLock(File)} or {@link #releaseSharedFileLock(File)} instead. + */ + @Deprecated + public static void releaseFileLock(@NotNull final File canonicalFile) { + releaseExclusiveFileLock(canonicalFile); + } + + private static void releaseFileLock0(@NotNull 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; + } + } + + ); + } + + /** + * Tries to execute a closure under exclusive file lock. + * If USE_LOCKING is false, provides synchronization only within local JVM. + * + * @param fileIOAction Closure to run, can throw {@link IOException}s. + * @return true if the lock was successfully acquired and IO action was executed, false otherwise. + */ + public static boolean tryRunExclusively(@NotNull final File canonicalFile, + @NotNull final FileChannel fileChannel, + @NotNull final FileIOAction fileIOAction) { + AtomicBoolean locked = new AtomicBoolean(false); + + FILE_LOCKS.compute(canonicalFile, (f, flr) -> { + if (flr != null) + return flr; + + try { + if (USE_EXCLUSIVE_LOCKING) { + try (FileLock ignored = fileChannel.tryLock()) { + if (ignored == null) + return null; + + fileIOAction.fileIOAction(); + + locked.set(true); + } + catch (OverlappingFileLockException ignored) { + // File lock is being held by this JVM, unsuccessful attempt. + } + } else { + fileIOAction.fileIOAction(); + + locked.set(true); + } + + return null; + } catch (Exception e) { + throw Jvm.rethrow(e); + } + } + ); + + return locked.get(); + } + /** * Executes a closure under exclusive file lock. * If USE_LOCKING is false, provides synchronization only within local JVM. @@ -104,7 +172,7 @@ public final class FileLockUtil { throw new ChronicleFileLockException("A file lock instance already exists for the file " + canonicalFile); try { - if (USE_LOCKING) { + if (USE_EXCLUSIVE_LOCKING) { try (FileLock ignored = fileChannel.lock()) { fileIOAction.fileIOAction(); } diff --git a/src/test/java/net/openhft/chronicle/map/FileLockUtilTest.java b/src/test/java/net/openhft/chronicle/map/FileLockUtilTest.java index 055eae48d..befb64ce8 100644 --- a/src/test/java/net/openhft/chronicle/map/FileLockUtilTest.java +++ b/src/test/java/net/openhft/chronicle/map/FileLockUtilTest.java @@ -13,8 +13,7 @@ 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; +import static org.junit.Assert.*; public class FileLockUtilTest { @@ -42,8 +41,8 @@ public class FileLockUtilTest { if (!OS.isWindows()) { FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); - FileLockUtil.releaseFileLock(canonicalFile); - FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseSharedFileLock(canonicalFile); + FileLockUtil.releaseSharedFileLock(canonicalFile); } } @@ -51,9 +50,9 @@ public class FileLockUtilTest { public void testExclusiveNormalCase() { if (!OS.isWindows()) { FileLockUtil.acquireExclusiveFileLock(canonicalFile, fileChannel); - FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseExclusiveFileLock(canonicalFile); FileLockUtil.acquireExclusiveFileLock(canonicalFile, fileChannel); - FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseExclusiveFileLock(canonicalFile); } } @@ -66,7 +65,7 @@ public class FileLockUtilTest { fail(); } catch (ChronicleFileLockException ignore) { } - FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseSharedFileLock(canonicalFile); } } @@ -79,7 +78,7 @@ public class FileLockUtilTest { fail(); } catch (ChronicleFileLockException ignore) { } - FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseExclusiveFileLock(canonicalFile); } } @@ -87,13 +86,13 @@ public class FileLockUtilTest { public void testComplicated() { if (!OS.isWindows()) { FileLockUtil.acquireExclusiveFileLock(canonicalFile, fileChannel); - FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseExclusiveFileLock(canonicalFile); FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); - FileLockUtil.releaseFileLock(canonicalFile); - FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseSharedFileLock(canonicalFile); + FileLockUtil.releaseSharedFileLock(canonicalFile); FileLockUtil.acquireExclusiveFileLock(canonicalFile, fileChannel); - FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseExclusiveFileLock(canonicalFile); } } @@ -115,8 +114,27 @@ public class FileLockUtilTest { }); fail(); } catch (ChronicleFileLockException e) { - FileLockUtil.releaseFileLock(canonicalFile); + FileLockUtil.releaseSharedFileLock(canonicalFile); } } } + + @Test + public void testTryRunExclusively() { + if (!OS.isWindows()) { + FileLockUtil.acquireSharedFileLock(canonicalFile, fileChannel); + + boolean lockedAndRun = FileLockUtil.tryRunExclusively(canonicalFile, fileChannel, () -> { + }); + + assertFalse(lockedAndRun); + + FileLockUtil.releaseSharedFileLock(canonicalFile); + + lockedAndRun = FileLockUtil.tryRunExclusively(canonicalFile, fileChannel, () -> { + }); + + assertTrue(lockedAndRun); + } + } } \ No newline at end of file