From 9e67840b4c3d022658dfff20f2ffc6f13430d8a2 Mon Sep 17 00:00:00 2001 From: Caleb Hulbert Date: Tue, 3 Mar 2026 11:21:12 -0500 Subject: [PATCH 1/8] refactor: make explicit the use of the FileLock in ChannelLock Signed-off-by: Caleb Hulbert --- src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java b/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java index 520a2b367..d97a53353 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java @@ -17,11 +17,14 @@ class ChannelLock implements Closeable { private final FileChannel channel; - private final FileLock lock; + + //Used to ensure a hard reference exists until we close + @SuppressWarnings({"unused", "FieldCanBeLocal"}) + private final FileLock unusedHardRef; private ChannelLock(final FileChannel channel, final FileLock lock) { this.channel = channel; - this.lock = lock; + this.unusedHardRef = lock; } public void close() throws IOException { From 5d994b5568a298700740ce1836ef132abbfdc598 Mon Sep 17 00:00:00 2001 From: tpietzsch Date: Thu, 5 Mar 2026 17:36:58 +0100 Subject: [PATCH 2/8] fix typos --- .../saalfeldlab/n5/cache/N5JsonCacheableContainer.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/main/java/org/janelia/saalfeldlab/n5/cache/N5JsonCacheableContainer.java b/src/main/java/org/janelia/saalfeldlab/n5/cache/N5JsonCacheableContainer.java index ba7c969a8..4edc181a2 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/cache/N5JsonCacheableContainer.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/cache/N5JsonCacheableContainer.java @@ -64,7 +64,7 @@ public interface N5JsonCacheableContainer { * the normalized path name * @param normalCacheKey * the normalized resource name (may be null). - * @return true if the resouce exists + * @return true if the resource exists */ boolean existsFromContainer(final String normalPathName, final String normalCacheKey); @@ -88,8 +88,8 @@ public interface N5JsonCacheableContainer { boolean isDatasetFromContainer(final String normalPathName); /** - * - * Returns true if a path is a group, given that the the given attributes exist + * + * Returns true if a path is a group, given that the given attributes exist * for the given cache key. *

* Should not call the backing storage. From a2bec03a6d7b7619edff3e89d0b942970d088a84 Mon Sep 17 00:00:00 2001 From: tpietzsch Date: Thu, 5 Mar 2026 14:14:04 +0100 Subject: [PATCH 3/8] Remove redundant modifiers --- src/main/java/org/janelia/saalfeldlab/n5/N5Reader.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/main/java/org/janelia/saalfeldlab/n5/N5Reader.java b/src/main/java/org/janelia/saalfeldlab/n5/N5Reader.java index aebbfa67d..742d401f3 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/N5Reader.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/N5Reader.java @@ -55,7 +55,7 @@ */ public interface N5Reader extends AutoCloseable { - public static class Version { + class Version { private final int major; private final int minor; @@ -192,17 +192,17 @@ public boolean isCompatible(final Version version) { /** * SemVer version of this N5 spec. */ - public static final Version NO_VERSION = new Version(0, 0, 0); + Version NO_VERSION = new Version(0, 0, 0); /** * SemVer version of this N5 spec. */ - public static final Version VERSION = new Version(4, 0, 0); + Version VERSION = new Version(4, 0, 0); /** * Version attribute key. */ - public static final String VERSION_KEY = "n5"; + String VERSION_KEY = "n5"; /** * Get the SemVer version of this container as specified in the 'version' From cad4cc20fb1968497187907ea3008f058e5a994c Mon Sep 17 00:00:00 2001 From: tpietzsch Date: Fri, 6 Mar 2026 08:48:29 +0100 Subject: [PATCH 4/8] fix wildcard imports --- .../n5/BufferedKvaLockedChannel.java | 20 +++++++++++++------ .../n5/FileSystemKeyValueAccess.java | 19 +++++++++++------- .../saalfeldlab/n5/codec/N5BlockCodecs.java | 2 +- 3 files changed, 27 insertions(+), 14 deletions(-) diff --git a/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java b/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java index 70ed4ba60..b6440b4bc 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java @@ -1,9 +1,17 @@ package org.janelia.saalfeldlab.n5; +import java.io.BufferedWriter; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.io.OutputStreamWriter; +import java.io.Reader; +import java.io.Writer; +import org.janelia.saalfeldlab.n5.N5Exception.N5IOException; import org.janelia.saalfeldlab.n5.readdata.ReadData; -import java.io.*; - class BufferedKvaLockedChannel implements LockedChannel { private final KeyValueAccess kva; @@ -16,24 +24,24 @@ class BufferedKvaLockedChannel implements LockedChannel { } @Override - public Reader newReader() throws N5Exception.N5IOException { + public Reader newReader() throws N5IOException { return new InputStreamReader(newInputStream()); } @Override - public InputStream newInputStream() throws N5Exception.N5IOException { + public InputStream newInputStream() throws N5IOException { return kva.createReadData(key).inputStream(); } @Override - public Writer newWriter() throws N5Exception.N5IOException { + public Writer newWriter() throws N5IOException { return new BufferedWriter(new OutputStreamWriter(newOutputStream())); } @Override - public OutputStream newOutputStream() throws N5Exception.N5IOException { + public OutputStream newOutputStream() throws N5IOException { if (baos == null) baos = new ByteArrayOutputStream(); return baos; diff --git a/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java b/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java index 83bb1c5ec..b1edae226 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java @@ -28,20 +28,25 @@ */ package org.janelia.saalfeldlab.n5; -import java.io.*; -import java.nio.file.*; - -import org.janelia.saalfeldlab.n5.N5Exception.N5IOException; -import org.janelia.saalfeldlab.n5.N5Exception.N5NoSuchKeyException; - +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; import java.net.URI; import java.net.URISyntaxException; +import java.nio.file.DirectoryNotEmptyException; +import java.nio.file.FileAlreadyExistsException; +import java.nio.file.FileSystemException; +import java.nio.file.Files; +import java.nio.file.NoSuchFileException; +import java.nio.file.Path; +import java.nio.file.Paths; import java.nio.file.attribute.FileAttribute; import java.util.Arrays; import java.util.Comparator; import java.util.Iterator; import java.util.stream.Stream; - +import org.janelia.saalfeldlab.n5.N5Exception.N5IOException; +import org.janelia.saalfeldlab.n5.N5Exception.N5NoSuchKeyException; import org.janelia.saalfeldlab.n5.readdata.ReadData; import org.janelia.saalfeldlab.n5.readdata.VolatileReadData; diff --git a/src/main/java/org/janelia/saalfeldlab/n5/codec/N5BlockCodecs.java b/src/main/java/org/janelia/saalfeldlab/n5/codec/N5BlockCodecs.java index db32bd114..6cb3aeadb 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/codec/N5BlockCodecs.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/codec/N5BlockCodecs.java @@ -47,7 +47,7 @@ import org.janelia.saalfeldlab.n5.StringDataBlock; import org.janelia.saalfeldlab.n5.readdata.ReadData; -import static org.janelia.saalfeldlab.n5.N5Exception.*; +import static org.janelia.saalfeldlab.n5.N5Exception.N5IOException; import static org.janelia.saalfeldlab.n5.codec.N5BlockCodecs.BlockHeader.MODE_DEFAULT; import static org.janelia.saalfeldlab.n5.codec.N5BlockCodecs.BlockHeader.MODE_OBJECT; import static org.janelia.saalfeldlab.n5.codec.N5BlockCodecs.BlockHeader.MODE_VARLENGTH; From 4776cc0115d4b2b106b72acb2e974318a12c4f7a Mon Sep 17 00:00:00 2001 From: tpietzsch Date: Fri, 6 Mar 2026 08:59:14 +0100 Subject: [PATCH 5/8] minimal javadoc --- .../org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java b/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java index b6440b4bc..617ec7524 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java @@ -12,6 +12,9 @@ import org.janelia.saalfeldlab.n5.N5Exception.N5IOException; import org.janelia.saalfeldlab.n5.readdata.ReadData; +/** + * Supports default implementation of the deprecated {@link KeyValueAccess#lockForReading(String)} and {@link KeyValueAccess#lockForWriting(String)} methods. + */ class BufferedKvaLockedChannel implements LockedChannel { private final KeyValueAccess kva; From 950174398d5fec58b6ad22d35377c3aed9df7ff5 Mon Sep 17 00:00:00 2001 From: tpietzsch Date: Fri, 6 Mar 2026 08:59:38 +0100 Subject: [PATCH 6/8] Add TODO @cmhulbert --- .../org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java b/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java index 617ec7524..e628cc211 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/BufferedKvaLockedChannel.java @@ -34,6 +34,9 @@ public Reader newReader() throws N5IOException { @Override public InputStream newInputStream() throws N5IOException { + + // TODO: This does not close the VolatileReadData returned by kva.createReadData. + // Easiest fix is probably to materialize the VolatileReadData and close it, then use the materialized readData. return kva.createReadData(key).inputStream(); } From 5cfda39cb1527b60d891de0cb60ec1dd4ad8bf3e Mon Sep 17 00:00:00 2001 From: tpietzsch Date: Fri, 6 Mar 2026 11:41:35 +0100 Subject: [PATCH 7/8] Remove tryLockForReading/Writing methods from FileKeyLockManager With respect to incorporating IoPolicy, the semantics of these methods are not clear. (Did we not get a lock because locking is broken for the file system, or because someone else already holds a write lock?) These methods were only used in tests. For simplicity, we just remove them for now. --- .../saalfeldlab/n5/FileKeyLockManager.java | 24 ------- .../janelia/saalfeldlab/n5/KeyLockState.java | 69 ------------------- .../n5/FileKeyLockManagerTest.java | 65 +---------------- 3 files changed, 3 insertions(+), 155 deletions(-) diff --git a/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java b/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java index b23b52ab8..7fbb0c7a2 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java @@ -135,30 +135,6 @@ public LockedFileChannel lockForWriting(final Path path) throws IOException { return keyLockState(path).acquireWrite(); } - /** - * Attempts to acquire a read lock for the specified key without blocking. - * - * @param path - * the file path to lock for reading - * @return a {@link LockedChannel} if the lock was acquired, null otherwise - */ - public LockedFileChannel tryLockForReading(final Path path) { - - return keyLockState(path).tryAcquireRead(); - } - - /** - * Attempts to acquire a write lock for the specified key without blocking. - * - * @param path - * the file path to lock for writing - * @return a {@link LockedChannel} if the lock was acquired, null otherwise - */ - public LockedFileChannel tryLockForWriting(final Path path) { - - return keyLockState(path).tryAcquireWrite(); - } - /** * Returns the number of keys currently being tracked. * diff --git a/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java b/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java index b002d6443..2701f7b75 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java @@ -85,56 +85,6 @@ LockedFileChannel acquireRead() throws IOException { } } - LockedFileChannel tryAcquireRead() { - - if(!readerMutex.tryAcquire()) { - return null; - } - - try { - if (numReaders == 0) { - // We are the first Reader, and are responsible for creating the channelLock - // (Other concurrent Readers are still blocked in readerMutex.) - - // If a Writer is still open, this will fail - if(!channelLockMutex.tryAcquire()) { - return null; - } - - try { - channelLock = ChannelLock.tryLock(path, false); - } catch (IOException e) { - // Something went wrong. Back off. - channelLockMutex.release(); - return null; - } - } - - // We have a READ ChannelLock. - // Try to open a FileChannel. - final FileChannel channel; - try { - channel = FileChannel.open(path, StandardOpenOption.READ); - } catch (final IOException e) { - // Something went wrong. Back off. - if (numReaders == 0) { - try { - releaseChannelLock(); - } catch (IOException ignored) { - } - } - return null; - } - - // We have a FileChannel. - // Create a LockedFileChannel that will releaseRead() when it is closed. - ++numReaders; - return new LockedFileChannel(channel, this::releaseRead); - } finally { - readerMutex.release(); - } - } - void releaseRead() throws IOException { try { @@ -184,25 +134,6 @@ LockedFileChannel acquireWrite() throws IOException { } } - LockedFileChannel tryAcquireWrite() { - - if (!channelLockMutex.tryAcquire()) { - return null; - } - - try { - channelLock = ChannelLock.tryLock(path, true); - } catch (IOException e) { - // Something went wrong. Back off. - channelLockMutex.release(); - return null; - } - - // We have a WRITE ChannelLock. - // Create a LockedFileChannel that will releaseWrite() when it is closed. - return new LockedFileChannel(channelLock.getChannel(), this::releaseWrite); - } - void releaseWrite() throws IOException { releaseChannelLock(); } diff --git a/src/test/java/org/janelia/saalfeldlab/n5/FileKeyLockManagerTest.java b/src/test/java/org/janelia/saalfeldlab/n5/FileKeyLockManagerTest.java index 49cf70cca..cceee34f9 100644 --- a/src/test/java/org/janelia/saalfeldlab/n5/FileKeyLockManagerTest.java +++ b/src/test/java/org/janelia/saalfeldlab/n5/FileKeyLockManagerTest.java @@ -33,6 +33,7 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import java.io.Closeable; @@ -40,6 +41,7 @@ import java.io.Reader; import java.io.Writer; import java.nio.file.Files; +import java.nio.file.NoSuchFileException; import java.nio.file.Path; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -194,66 +196,6 @@ public void testReadWriteExclusion() throws Exception { assertEquals("Reader should see written content", writtenContent, readContent.get()); } - @Test - public void testTryLock() throws Exception { - - /* create a test file */ - final Path testFile = tempDir.resolve("test3.txt"); - final String testContent = "test content"; - Files.write(testFile, testContent.getBytes()); - - /* try acquiring read lock and read */ - LockedChannel readLock1 = FILE_LOCK_MANAGER.tryLockForReading(testFile); - assertNotNull("Should acquire read lock", readLock1); - try (final Reader reader = readLock1.newReader()) { - final char[] buf = new char[testContent.length()]; - assertEquals("Should read correct number of chars", testContent.length(), reader.read(buf)); - } - - /* try acquiring another read lock and read */ - LockedChannel readLock2 = FILE_LOCK_MANAGER.tryLockForReading(testFile); - assertNotNull("Should acquire second read lock", readLock2); - try (final Reader reader = readLock2.newReader()) { - final char[] buf = new char[testContent.length()]; - assertEquals("Should read correct number of chars", testContent.length(), reader.read(buf)); - } - - /* try acquiring write lock while reads are held - should fail */ - LockedChannel writeLock = FILE_LOCK_MANAGER.tryLockForWriting(testFile); - assertNull("Should not acquire write lock while reads are held", writeLock); - - readLock1.close(); - readLock2.close(); - - /* now try write lock and write */ - writeLock = FILE_LOCK_MANAGER.tryLockForWriting(testFile); - assertNotNull("Should acquire write lock after reads are released", writeLock); - final String newContent = "new content"; - try (final Writer writer = writeLock.newWriter()) { - writer.write(newContent); - } - - /* try acquiring read lock from another thread while write is held */ - final AtomicReference readLock3 = new AtomicReference<>(); - final Thread readerThread = new Thread(() -> { - readLock3.set(FILE_LOCK_MANAGER.tryLockForReading(testFile)); - }); - readerThread.start(); - readerThread.join(); - - assertNull("Should not acquire read lock while write is held", readLock3.get()); - - writeLock.close(); - - /* verify written content */ - try (final LockedChannel verifyLock = FILE_LOCK_MANAGER.lockForReading(testFile); - final Reader reader = verifyLock.newReader()) { - final char[] buf = new char[newContent.length()]; - reader.read(buf); - assertEquals("Content should match what was written", newContent, new String(buf)); - } - } - private class CleanUpHelper implements Closeable { private final Path path; @@ -391,8 +333,7 @@ public void testWriteLockCreatesParentDirectories() throws Exception { @Test public void testReadLockRequiresExistingFile() throws Exception { final Path testFile = tempDir.resolve("nonexistent.txt"); - assertNull("Should not acquire read lock for non-existent file", - FILE_LOCK_MANAGER.tryLockForReading(testFile)); + assertThrows("Should not acquire read lock for non-existent file", NoSuchFileException.class, () -> FILE_LOCK_MANAGER.lockForReading(testFile)); } @Test From e709b59fc7307a91686ad0d28166b308e98cd54a Mon Sep 17 00:00:00 2001 From: tpietzsch Date: Fri, 6 Mar 2026 10:25:20 +0100 Subject: [PATCH 8/8] Rework IoPolicy Make IoPolicy an enum that just communicates the intented locking behaviour (STRICT, PERMISSIVE, UNSAFE). How exactly this is implemented is up to the backend. For the FileSystemKeyValueAccess, the IoPolicy is simply to the FileKeyLockManager, then on to the FileLockState, and finally the ChannelLock, which actually implements the behaviour using FileLock (or not). --- .../janelia/saalfeldlab/n5/ChannelLock.java | 54 +++--- .../saalfeldlab/n5/FileKeyLockManager.java | 88 ++++++--- .../n5/FileSystemKeyValueAccess.java | 177 ++++++++++++++---- .../janelia/saalfeldlab/n5/FsIoPolicy.java | 148 --------------- .../org/janelia/saalfeldlab/n5/IoPolicy.java | 95 +++++----- .../janelia/saalfeldlab/n5/KeyLockState.java | 13 +- 6 files changed, 285 insertions(+), 290 deletions(-) delete mode 100644 src/main/java/org/janelia/saalfeldlab/n5/FsIoPolicy.java diff --git a/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java b/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java index d97a53353..c51c9f9db 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java @@ -18,13 +18,22 @@ class ChannelLock implements Closeable { private final FileChannel channel; - //Used to ensure a hard reference exists until we close + /** + * Hold a hard reference to the {@code FileLock} to make sure it is not + * prematurely released. + *

+ * NB: We do not call {@code lock.release()} in {@link #close}, because at + * this point the channel might be already closed (by an external writer). + * {@code lock.release()} will throw an exception if the channel is already + * closed. Instead, we just close the channel which will automatically + * release the lock. + */ @SuppressWarnings({"unused", "FieldCanBeLocal"}) - private final FileLock unusedHardRef; + private final FileLock lock; private ChannelLock(final FileChannel channel, final FileLock lock) { this.channel = channel; - this.unusedHardRef = lock; + this.lock = lock; } public void close() throws IOException { @@ -56,9 +65,12 @@ FileChannel getChannel() { * @throws IOException if an error occurs while opening the channel, or if * the calling thread is interrupted while waiting for the {@code FileLock}. */ - static ChannelLock lock(final Path path, final boolean forWriting) throws IOException { + static ChannelLock lock(final Path path, final boolean forWriting, final IoPolicy policy) throws IOException { final FileChannel channel = openFileChannel(path, forWriting); + if (policy == IoPolicy.UNSAFE) { + return new ChannelLock(channel, null); + } try { while (true) { try { @@ -74,34 +86,12 @@ static ChannelLock lock(final Path path, final boolean forWriting) throws IOExce } } } catch (Exception e) { - closeQuietly(channel); - throw e; - } - } - - /** - * Create a {@link FileChannel} on the given {@code path} and try to lock it - * with a system-level {@link FileLock}. If the channel cannot be locked, - * {@code null} is returned. - *

- * The {@code FileLock} is exclusive if the {@code path} is locked {@code - * forWriting}, and shared otherwise. - *

- * If the {@code path} is locked {@code forWriting} non-existing file and - * the parent directories are created as needed. - * - * @throws IOException if an error occurs while opening the channel. - */ - static ChannelLock tryLock(final Path path, final boolean forWriting) throws IOException { - - FileChannel channel = null; - try { - channel = openFileChannel(path, forWriting); - final FileLock lock = channel.tryLock(0, Long.MAX_VALUE, !forWriting); - return lock == null ? null : new ChannelLock(channel, lock); - } catch (Exception e) { - closeQuietly(channel); - throw e; + if (policy == IoPolicy.STRICT) { + closeQuietly(channel); + throw e; + } else { + return new ChannelLock(channel, null); + } } } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java b/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java index 7fbb0c7a2..977832de2 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java @@ -31,6 +31,7 @@ import java.io.IOException; import java.lang.ref.ReferenceQueue; import java.lang.ref.WeakReference; +import java.nio.channels.FileLock; import java.nio.file.Path; import java.util.concurrent.ConcurrentHashMap; @@ -79,7 +80,7 @@ private void cleanUp() } } - private KeyLockState keyLockState(final Path path) { + private KeyLockState keyLockState(final Path path, final IoPolicy policy) throws IOException { final String key = path.toAbsolutePath().toString(); @@ -87,17 +88,18 @@ private KeyLockState keyLockState(final Path path) { final WeakValue existingRef = locks.get(key); KeyLockState state = existingRef == null ? null : existingRef.get(); - if (state != null) { - return state; + if (state == null) { + final KeyLockState newState = new KeyLockState(path, policy); + while (state == null) { + final WeakValue ref = locks.compute(key, + (k, v) -> (v != null && v.get() != null) + ? v + : new WeakValue(k, newState, refQueue)); + state = ref.get(); + } } - - final KeyLockState newState = new KeyLockState(path); - while (state == null) { - final WeakValue ref = locks.compute(key, - (k, v) -> (v != null && v.get() != null) - ? v - : new WeakValue(k, newState, refQueue)); - state = ref.get(); + if (state.policy() != policy) { + throw new IOException("Trying to lock \"" + path + "\" with policy " + policy + ", but it is already used with " + state.policy()); } return state; } @@ -108,31 +110,73 @@ private KeyLockState keyLockState(final Path path) { *

* The first reader will acquire a shared file lock. Subsequent readers * only acquire the thread-level lock. + *

+ * The given locking {@link IoPolicy policy} applies to OS-level locking. + * For both the {@code STRICT} and {@code PERMISSIVE} policy, a {@link + * FileLock} is obtained. If this fails, {@code STRICT} will throw an {@code + * IOException}. {@code PERMISSIVE} will proceed without locking. {@code + * UNSAFE} will not attempt OS-level locking, however will still manage + * mutual exclusion of readers and writers in the same JVM. Trying to lock + * the same path with different locking policies will throw an {@code + * IOException}. * * @param path - * the key (file path) to lock for reading + * the key (file path) to lock for reading + * @param policy + * the locking policy + * * @return a {@link LockedChannel} that must be closed when done + * * @throws IOException - * if acquiring the file lock fails + * if acquiring the file lock fails + */ + public LockedFileChannel lockForReading(final Path path, final IoPolicy policy) throws IOException { + + return keyLockState(path, policy).acquireRead(); + } + + /** + * Acquires a read lock for the specified key with the {@link IoPolicy#STRICT} locking policy. */ public LockedFileChannel lockForReading(final Path path) throws IOException { - return keyLockState(path).acquireRead(); + return lockForReading(path, IoPolicy.STRICT); } /** * Acquires a write lock for the specified key. Only one thread can hold a * write lock for a key at a time, and no readers can hold locks. + *

+ * The given locking {@link IoPolicy policy} applies to OS-level locking. + * For both the {@code STRICT} and {@code PERMISSIVE} policy, a {@link + * FileLock} is obtained. If this fails, {@code STRICT} will throw an {@code + * IOException}. {@code PERMISSIVE} will proceed without locking. {@code + * UNSAFE} will not attempt OS-level locking, however will still manage + * mutual exclusion of readers and writers in the same JVM. Trying to lock + * the same path with different locking policies will throw an {@code + * IOException}. * * @param path - * the file path to lock for writing + * the file path to lock for writing + * @param policy + * the locking policy + * * @return a {@link LockedChannel} that must be closed when done + * * @throws IOException - * if acquiring the file lock fails + * if acquiring the file lock fails + */ + public LockedFileChannel lockForWriting(final Path path, final IoPolicy policy) throws IOException { + + return keyLockState(path, policy).acquireWrite(); + } + + /** + * Acquires a write lock for the specified key with the {@link IoPolicy#STRICT} locking policy. */ public LockedFileChannel lockForWriting(final Path path) throws IOException { - return keyLockState(path).acquireWrite(); + return lockForWriting(path, IoPolicy.STRICT); } /** @@ -144,14 +188,4 @@ int size() { return locks.size(); } - - /** - * Removes the lock state for a key if no locks are held. - * - * @param key the key whose lock state should be removed - * @return true if removed, false if currently in use or not found - */ - public boolean removeLockIfUnused(final String key) { - throw new UnsupportedOperationException("TODO. REMOVE"); - } } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java b/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java index b1edae226..d1f9da1f8 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java @@ -33,6 +33,8 @@ import java.io.UncheckedIOException; import java.net.URI; import java.net.URISyntaxException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; import java.nio.file.DirectoryNotEmptyException; import java.nio.file.FileAlreadyExistsException; import java.nio.file.FileSystemException; @@ -40,6 +42,7 @@ import java.nio.file.NoSuchFileException; import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; import java.nio.file.attribute.FileAttribute; import java.util.Arrays; import java.util.Comparator; @@ -47,9 +50,12 @@ import java.util.stream.Stream; import org.janelia.saalfeldlab.n5.N5Exception.N5IOException; import org.janelia.saalfeldlab.n5.N5Exception.N5NoSuchKeyException; +import org.janelia.saalfeldlab.n5.readdata.LazyRead; import org.janelia.saalfeldlab.n5.readdata.ReadData; import org.janelia.saalfeldlab.n5.readdata.VolatileReadData; +import static org.janelia.saalfeldlab.n5.FileKeyLockManager.FILE_LOCK_MANAGER; + /** * Filesystem {@link KeyValueAccess}. * @@ -59,44 +65,65 @@ */ public class FileSystemKeyValueAccess implements KeyValueAccess { - private static final IoPolicy ioPolicy = getIoPolicy(); + private final IoPolicy ioPolicy; + + /** + * Create a {@link FileSystemKeyValueAccess}. + */ + public FileSystemKeyValueAccess() { - private static IoPolicy getIoPolicy() { String property = System.getProperty("n5.ioPolicy"); - if (property == null) - return FsIoPolicy.atomicWithFallback; - - switch (property) { - case "strict": - return new FsIoPolicy.Atomic(); - case "unsafe": - return new FsIoPolicy.Unsafe(); - case "permissive": - default: - return FsIoPolicy.atomicWithFallback; + ioPolicy = IoPolicy.fromString(property); + } + + @Override + public VolatileReadData createReadData(final String normalPath) { + return VolatileReadData.from(new FileLazyRead(Paths.get(normalPath), ioPolicy)); + } + + @Override + public void write(final String normalPath, final ReadData data) throws N5IOException { + final Path path = Paths.get(normalPath); + try (final LockedFileChannel channel = lockForWriting(path, ioPolicy)) { + data.writeTo(channel.newOutputStream()); + } catch (IOException e) { + throw new N5IOException(e); } } + @Deprecated @Override - public VolatileReadData createReadData(final String normalPath) throws N5IOException { + public LockedFileChannel lockForReading(final String normalPath) throws N5IOException { + + return lockForReading(Paths.get(normalPath), ioPolicy); + } + + @Deprecated + @Override + public LockedChannel lockForWriting(final String normalPath) throws N5IOException { + + return lockForWriting(Paths.get(normalPath), ioPolicy); + } + + private static LockedFileChannel lockForReading(final Path path, final IoPolicy policy) throws N5IOException { try { - return ioPolicy.read(normalPath); + return FILE_LOCK_MANAGER.lockForReading(path, policy); } catch (final NoSuchFileException e) { throw new N5NoSuchKeyException("No such file", e); } catch (IOException | UncheckedIOException e) { - throw new N5IOException("Failed to lock file for reading: " + normalPath, e); + throw new N5IOException("Failed to lock file for reading: " + path, e); } } - @Override - public void write(final String normalPath, final ReadData data) throws N5IOException { + private LockedFileChannel lockForWriting(final Path path, final IoPolicy policy) throws N5IOException { + try { - ioPolicy.write(normalPath, data); + return FILE_LOCK_MANAGER.lockForWriting(path, policy); } catch (final NoSuchFileException e) { throw new N5NoSuchKeyException("No such file", e); } catch (IOException | UncheckedIOException e) { - throw new N5IOException("Failed to lock file for writing: " + normalPath, e); + throw new N5IOException("Failed to lock file for writing: " + path, e); } } @@ -127,6 +154,17 @@ public long size(final String normalPath) { return size(Paths.get(normalPath)); } + private static long size(final Path path) { + + try { + return Files.size(path); + } catch (NoSuchFileException e) { + throw new N5NoSuchKeyException("No such file", e); + } catch (IOException | UncheckedIOException e) { + throw new N5IOException(e); + } + } + @Override public String[] listDirectories(final String normalPath) throws N5IOException { @@ -278,13 +316,17 @@ public void delete(final String normalPath) throws N5IOException { final Path path = Paths.get(normalPath); if (Files.isRegularFile(path)) - ioPolicy.delete(normalPath); + try (final LockedChannel channel = lockForWriting(path, ioPolicy)) { + Files.delete(path); + } else { try (final Stream pathStream = Files.walk(path)) { - for (final Iterator i = pathStream.sorted(Comparator.reverseOrder()).iterator(); i.hasNext(); ) { + for (final Iterator i = pathStream.sorted(Comparator.reverseOrder()).iterator(); i.hasNext();) { final Path childPath = i.next(); if (Files.isRegularFile(childPath)) - ioPolicy.delete(childPath.toString()); + try (final LockedChannel channel = lockForWriting(childPath, ioPolicy)) { + Files.delete(childPath); + } else tryDelete(childPath); } @@ -297,21 +339,10 @@ public void delete(final String normalPath) throws N5IOException { } } - protected static long size(final Path path) { - - try { - return Files.size(path); - } catch (NoSuchFileException e) { - throw new N5NoSuchKeyException("No such file", e); - } catch (IOException | UncheckedIOException e) { - throw new N5IOException(e); - } - } - protected static void tryDelete(final Path path) throws IOException { try { - ioPolicy.delete(path.toString()); + Files.delete(path); } catch (final DirectoryNotEmptyException e) { /* * Even though path is expected to be an empty directory, sometimes @@ -321,7 +352,7 @@ protected static void tryDelete(final Path path) throws IOException { try { /* wait and reattempt */ Thread.sleep(100); - ioPolicy.delete(path.toString()); + Files.delete(path); } catch (final InterruptedException ex) { e.printStackTrace(); Thread.currentThread().interrupt(); @@ -463,4 +494,78 @@ protected static void createAndCheckIsDirectory( throw x; } } + + private static class FileLazyRead implements LazyRead { + + private final Path path; + private LockedFileChannel lock; + + FileLazyRead(final Path path, final IoPolicy policy) { + this.path = path; + lock = lockForReading(path, policy); + } + + @Override + public long size() throws N5IOException { + + if (lock == null) { + throw new N5IOException("FileLazyRead is already closed."); + } + return FileSystemKeyValueAccess.size(path); + } + + @Override + public ReadData materialize(final long offset, final long length) { + + if (lock == null) { + throw new N5IOException("FileLazyRead is already closed."); + } + + try (final FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) { + + channel.position(offset); + + final long channelSize = channel.size(); + if (!validBounds(channelSize, offset, length)) { + throw new IndexOutOfBoundsException(); + } + + final long size = length < 0 ? (channelSize - offset) : length; + if (size > Integer.MAX_VALUE) { + throw new IndexOutOfBoundsException("Attempt to materialize too large data"); + } + + final byte[] data = new byte[(int) size]; + final ByteBuffer buf = ByteBuffer.wrap(data); + channel.read(buf); + return ReadData.from(data); + + } catch (final NoSuchFileException e) { + throw new N5NoSuchKeyException("No such file", e); + } catch (IOException | UncheckedIOException e) { + throw new N5IOException(e); + } + } + + @Override + public void close() throws IOException { + + if (lock != null) { + lock.close(); + lock = null; + } + } + + private static boolean validBounds(long channelSize, long offset, long length) { + + if (offset < 0) + return false; + else if (channelSize > 0 && offset >= channelSize) // offset == 0 and channelSize == 0 is okay + return false; + else if (length >= 0 && offset + length > channelSize) + return false; + + return true; + } + } } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/FsIoPolicy.java b/src/main/java/org/janelia/saalfeldlab/n5/FsIoPolicy.java deleted file mode 100644 index 5d01dd919..000000000 --- a/src/main/java/org/janelia/saalfeldlab/n5/FsIoPolicy.java +++ /dev/null @@ -1,148 +0,0 @@ -package org.janelia.saalfeldlab.n5; - -import org.janelia.saalfeldlab.n5.readdata.LazyRead; -import org.janelia.saalfeldlab.n5.readdata.ReadData; -import org.janelia.saalfeldlab.n5.readdata.VolatileReadData; - -import java.io.Closeable; -import java.io.IOException; -import java.io.UncheckedIOException; -import java.nio.ByteBuffer; -import java.nio.channels.FileChannel; -import java.nio.file.*; - -import static org.janelia.saalfeldlab.n5.FileKeyLockManager.FILE_LOCK_MANAGER; - -public class FsIoPolicy { - - static final IoPolicy atomicWithFallback = IoPolicy.withFallback(new Atomic(), new Unsafe()); - - private static boolean validBounds(long channelSize, long offset, long length) { - - if (offset < 0) - return false; - else if (channelSize > 0 && offset >= channelSize) // offset == 0 and channelSize == 0 is okay - return false; - else if (length >= 0 && offset + length > channelSize) - return false; - - return true; - } - - public static class Unsafe implements IoPolicy { - @Override - public void write(String key, ReadData readData) throws IOException { - final Path path = Paths.get(key); - Files.copy(readData.inputStream(), path, StandardCopyOption.REPLACE_EXISTING); - } - - @Override - public VolatileReadData read(final String key) throws IOException { - final Path path = Paths.get(key); - FileLazyRead fileLazyRead = new FileLazyRead(path, false); - return VolatileReadData.from(fileLazyRead); - } - - @Override - public void delete(final String key) throws IOException { - final Path path = Paths.get(key); - Files.deleteIfExists(path); - } - } - - public static class Atomic implements IoPolicy { - @Override - public void write(String key, ReadData readData) throws IOException { - final Path path = Paths.get(key); - try (LockedFileChannel channel = FILE_LOCK_MANAGER.lockForWriting(path)) { - readData.writeTo(channel.newOutputStream()); - } - } - - @Override - public VolatileReadData read(String key) throws IOException { - final Path path = Paths.get(key); - FileLazyRead fileLazyRead = new FileLazyRead(path, true); - return VolatileReadData.from(fileLazyRead); - } - - @Override - public void delete(final String key) throws IOException { - final Path path = Paths.get(key); - try (LockedFileChannel ignore = FILE_LOCK_MANAGER.lockForWriting(path)) { - Files.delete(path); - } - } - } - - static class FileLazyRead implements LazyRead { - - private static final Closeable NO_OP = () -> { }; - - private final Path path; - private Closeable lock; - - FileLazyRead(final Path path) throws IOException { - this(path, true); - } - - FileLazyRead(final Path path, final boolean requireLock ) throws IOException { - this.path = path; - if (requireLock) - lock = FILE_LOCK_MANAGER.lockForReading(path); - else - lock = NO_OP; - } - - @Override - public long size() throws N5Exception.N5IOException { - - if (lock == null) { - throw new N5Exception.N5IOException("FileLazyRead is already closed."); - } - return FileSystemKeyValueAccess.size(path); - } - - @Override - public ReadData materialize(final long offset, final long length) { - - if (lock == null) { - throw new N5Exception.N5IOException("FileLazyRead is already closed."); - } - - try (final FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) { - - channel.position(offset); - - final long channelSize = channel.size(); - if (!validBounds(channelSize, offset, length)) { - throw new IndexOutOfBoundsException(); - } - - final long size = length < 0 ? (channelSize - offset) : length; - if (size > Integer.MAX_VALUE) { - throw new IndexOutOfBoundsException("Attempt to materialize too large data"); - } - - final byte[] data = new byte[(int) size]; - final ByteBuffer buf = ByteBuffer.wrap(data); - channel.read(buf); - return ReadData.from(data); - - } catch (final NoSuchFileException e) { - throw new N5Exception.N5NoSuchKeyException("No such file", e); - } catch (IOException | UncheckedIOException e) { - throw new N5Exception.N5IOException(e); - } - } - - @Override - public void close() throws IOException { - - if (lock != null) { - lock.close(); - lock = null; - } - } - } -} diff --git a/src/main/java/org/janelia/saalfeldlab/n5/IoPolicy.java b/src/main/java/org/janelia/saalfeldlab/n5/IoPolicy.java index b1dcb3044..40f34a23c 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/IoPolicy.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/IoPolicy.java @@ -1,47 +1,54 @@ package org.janelia.saalfeldlab.n5; -import org.janelia.saalfeldlab.n5.readdata.ReadData; -import org.janelia.saalfeldlab.n5.readdata.VolatileReadData; - -import java.io.IOException; - -public interface IoPolicy { - - void write(String key, ReadData readData) throws IOException; - - VolatileReadData read(String key) throws IOException; - - void delete(String key) throws IOException; - - static IoPolicy withFallback(IoPolicy primary, IoPolicy fallback) { - return new IoPolicy() { - - @Override - public void write(String key, ReadData readData) throws IOException { - try { - primary.write(key, readData); - } catch (IOException e) { - fallback.write(key, readData); - } - } - - @Override - public VolatileReadData read(String key) throws IOException { - try { - return primary.read(key); - } catch (IOException e) { - return fallback.read(key); - } - } - - @Override - public void delete(String key) throws IOException { - try { - primary.delete(key); - } catch (IOException e) { - fallback.delete(key); - } - } - }; - } +/** + * File locking policy. + *

+ * Usually, we want to coordinate reads and writs to a container such that + *

    + *
  • multiple readers can access a key simultaneously (blocking all writers).
    • + *
  • A writer should have exclusive access to a key (blocking all other readers and writers).
    • + *
+ * However, this cannot always be enforced for all backends: + * For SMB on macOS OS-level file locking is broken. + * For AWS S3 and Google Cloud we can detect (but not prevent) concurrent modifications. + *

+ * Sometimes we know that we can disregard locking, for example in read-only settings. + *

+ * {@code IoPolicy} can be used to configure locking policy for backends ({@link + * KeyValueAccess}) that support various locking strategies (potentially coming + * with performance trade-offs). + *

+ * The policy values ({@link #STRICT}, {@link #UNSAFE}, {@link #PERMISSIVE}) + * specify intent. Detailed interpretation is up to the backend implementation. + */ +public enum IoPolicy { + + /** + * Protect all reads and writes by locks. + * Fail if locking is not possible. + */ + STRICT, + + /** + * Reads and writes are unprotected. + */ + UNSAFE, + + /** + * Try to lock for all reads and writes. + * Fall back to unprotected reads and writes if locking is not possible. + * This is the default. + */ + PERMISSIVE; + + static IoPolicy fromString(final String s) { + if ("strict".equalsIgnoreCase(s)) + return STRICT; + else if ("unsafe".equalsIgnoreCase(s)) + return UNSAFE; +// else if ("permissive".equalsIgnoreCase(s)) +// return PERMISSIVE; + else + return PERMISSIVE; + } } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java b/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java index 2701f7b75..e00620f63 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java @@ -13,8 +13,11 @@ class KeyLockState { private final Path path; - public KeyLockState(final Path path) { + private final IoPolicy policy; + + public KeyLockState(final Path path, IoPolicy policy) { this.path = path; + this.policy = policy; } /** @@ -52,7 +55,7 @@ LockedFileChannel acquireRead() throws IOException { channelLockMutex.acquire(); try { - channelLock = ChannelLock.lock(path, false); + channelLock = ChannelLock.lock(path, false, policy); } catch (IOException e) { // Something went wrong. Back off. channelLockMutex.release(); @@ -119,7 +122,7 @@ LockedFileChannel acquireWrite() throws IOException { channelLockMutex.acquire(); try { - channelLock = ChannelLock.lock(path, true); + channelLock = ChannelLock.lock(path, true, policy); } catch (IOException e) { // Something went wrong. Back off. channelLockMutex.release(); @@ -137,4 +140,8 @@ LockedFileChannel acquireWrite() throws IOException { void releaseWrite() throws IOException { releaseChannelLock(); } + + IoPolicy policy() { + return policy; + } }