diff --git a/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java b/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java new file mode 100644 index 000000000..520a2b367 --- /dev/null +++ b/src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java @@ -0,0 +1,133 @@ +package org.janelia.saalfeldlab.n5; + +import java.io.Closeable; +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.channels.FileLock; +import java.nio.channels.OverlappingFileLockException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; + +/** + * Holds a channel and system-level file lock (shared for writing, non-shared + * for reading) and keeps it open until this {@code ChannelLock} is {@link + * #close() closed}. + */ +class ChannelLock implements Closeable { + + private final FileChannel channel; + private final FileLock lock; + + private ChannelLock(final FileChannel channel, final FileLock lock) { + this.channel = channel; + this.lock = lock; + } + + public void close() throws IOException { + + // NB: We do not call lock.release() here, because it may throw an + // exception if the channel is already closed. Instead, we just close + // the channel. This will automatically release the lock. (And it is ok + // to close an already closed channel.) + + channel.close(); + } + + FileChannel getChannel() { + return channel; + } + + /** + * Create a {@link FileChannel} on the given {@code path} and lock it with a + * system-level {@link FileLock}. If there is an existing overlapping file + * lock, this method will block until the existing lock is released and the + * channel could be locked (by us). + *

+ * 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, or if + * the calling thread is interrupted while waiting for the {@code FileLock}. + */ + static ChannelLock lock(final Path path, final boolean forWriting) throws IOException { + + final FileChannel channel = openFileChannel(path, forWriting); + try { + while (true) { + try { + final FileLock lock = channel.lock(0, Long.MAX_VALUE, !forWriting); + return new ChannelLock(channel, lock); + } catch (final OverlappingFileLockException e) { + try { + Thread.sleep(100); + } catch (final InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while waiting for file lock", ie); + } + } + } + } 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; + } + } + + /** + * Opens a file channel. If the channel is opened {@code forWriting}, + * then this may create the file and the parent directories as needed. + * + * @throws IOException + * if the channel cannot be opened + */ + private static FileChannel openFileChannel(final Path path, final boolean forWriting) throws IOException { + + if (forWriting) { + final Path parent = path.getParent(); + if (parent != null) { + Files.createDirectories(parent); + } + return FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.CREATE); + } else { + return FileChannel.open(path, StandardOpenOption.READ); + } + } + + private static void closeQuietly(final FileChannel fileChannel) { + if (fileChannel != null) { + try { + fileChannel.close(); + } catch (final IOException ignored) { + } + } + } +} diff --git a/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java b/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java new file mode 100644 index 000000000..b23b52ab8 --- /dev/null +++ b/src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java @@ -0,0 +1,181 @@ +/*- + * #%L + * Not HDF5 + * %% + * Copyright (C) 2017 - 2025 Stephan Saalfeld + * %% + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright notice, + * this list of conditions and the following disclaimer in the documentation + * and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + * #L% + */ +package org.janelia.saalfeldlab.n5; + +import java.io.IOException; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.nio.file.Path; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Provides thread-safe and process-safe read/write locking for filesystem paths. + * Uses thread locks for JVM coordination and file locks for inter-process coordination. + */ +class FileKeyLockManager { + + static final FileKeyLockManager FILE_LOCK_MANAGER = new FileKeyLockManager(); + + private FileKeyLockManager() { + // singleton + } + + + private final ConcurrentHashMap locks = new ConcurrentHashMap<>(); + + private final ReferenceQueue refQueue = new ReferenceQueue<>(); + + private static class WeakValue extends WeakReference { + + final String key; + + WeakValue( + final String key, + final KeyLockState value, + final ReferenceQueue queue) { + + super(value, queue); + this.key = key; + } + } + + /** + * Remove entries from the cache whose references have been + * garbage-collected. + */ + private void cleanUp() + { + while (true) { + final WeakValue ref = (WeakValue) refQueue.poll(); + if (ref == null) + break; + locks.remove(ref.key, ref); + } + } + + private KeyLockState keyLockState(final Path path) { + + final String key = path.toAbsolutePath().toString(); + + cleanUp(); + + final WeakValue existingRef = locks.get(key); + KeyLockState state = existingRef == null ? null : existingRef.get(); + if (state != null) { + return state; + } + + 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(); + } + return state; + } + + /** + * Acquires a read lock for the specified key. Multiple threads can hold + * read locks for the same key simultaneously. + *

+ * The first reader will acquire a shared file lock. Subsequent readers + * only acquire the thread-level lock. + * + * @param path + * the key (file path) to lock for reading + * @return a {@link LockedChannel} that must be closed when done + * @throws IOException + * if acquiring the file lock fails + */ + public LockedFileChannel lockForReading(final Path path) throws IOException { + + return keyLockState(path).acquireRead(); + } + + /** + * 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. + * + * @param path + * the file path to lock for writing + * @return a {@link LockedChannel} that must be closed when done + * @throws IOException + * if acquiring the file lock fails + */ + 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. + * + * @return the number of keys with associated locks + */ + 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 e6ae8439b..378b1466f 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/FileSystemKeyValueAccess.java @@ -33,28 +33,19 @@ import java.io.File; import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.io.Reader; import java.io.UncheckedIOException; -import java.io.Writer; -import java.net.URI; -import java.net.URISyntaxException; import java.nio.ByteBuffer; -import java.nio.channels.Channels; import java.nio.channels.FileChannel; -import java.nio.channels.OverlappingFileLockException; -import java.nio.charset.StandardCharsets; +import java.net.URI; +import java.net.URISyntaxException; import java.nio.file.DirectoryNotEmptyException; import java.nio.file.FileAlreadyExistsException; import java.nio.file.FileSystem; import java.nio.file.FileSystemException; import java.nio.file.Files; import java.nio.file.NoSuchFileException; -import java.nio.file.OpenOption; 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; @@ -63,6 +54,8 @@ import org.janelia.saalfeldlab.n5.readdata.ReadData; +import static org.janelia.saalfeldlab.n5.FileKeyLockManager.FILE_LOCK_MANAGER; + /** * Filesystem {@link KeyValueAccess}. * @@ -72,105 +65,6 @@ */ public class FileSystemKeyValueAccess implements KeyValueAccess { - /** - * A {@link FileChannel} wrapper that attempts to acquire a lock and waits - * for existing locks to be lifted before returning if the - * {@link FileSystem} supports that. If the {@link FileSystem} does not - * support locking, it returns immediately. - */ - protected class LockedFileChannel implements LockedChannel { - - protected final FileChannel channel; - - protected LockedFileChannel(final String path, final boolean readOnly) throws IOException { - - this(fileSystem.getPath(path), readOnly); - } - - protected LockedFileChannel(final Path path, final boolean readOnly) throws IOException { - - final OpenOption[] options; - if (readOnly) { - options = new OpenOption[]{StandardOpenOption.READ}; - channel = FileChannel.open(path, options); - } else { - options = new OpenOption[]{StandardOpenOption.READ, StandardOpenOption.WRITE, - StandardOpenOption.CREATE}; - FileChannel tryChannel = null; - try { - tryChannel = FileChannel.open(path, options); - } catch (final NoSuchFileException e) { - createDirectories(path.getParent()); - tryChannel = FileChannel.open(path, options); - } finally { - channel = tryChannel; - } - } - - for (boolean waiting = true; waiting;) { - waiting = false; - try { - channel.lock(0L, Long.MAX_VALUE, readOnly); - } catch (final OverlappingFileLockException e) { - waiting = true; - try { - Thread.sleep(100); - } catch (final InterruptedException f) { - waiting = false; - Thread.currentThread().interrupt(); - } - } catch (final IOException e) {} - } - } - - protected FileChannel getFileChannel() { - return channel; - } - - private void truncateChannel(int size) { - - try { - channel.truncate(size); - } catch (NoSuchFileException e) { - throw new N5NoSuchKeyException("No such file", e); - } catch (IOException | UncheckedIOException e ) { - throw new N5IOException("Failed to truncate locked file channel", e); - } - } - - @Override - public Reader newReader() throws N5IOException { - - return Channels.newReader(channel, StandardCharsets.UTF_8.name()); - } - - @Override - public Writer newWriter() throws N5IOException { - - truncateChannel(0); - return Channels.newWriter(channel, StandardCharsets.UTF_8.name()); - } - - @Override - public InputStream newInputStream() throws N5IOException { - - return Channels.newInputStream(channel); - } - - @Override - public OutputStream newOutputStream() throws N5IOException { - - truncateChannel(0); - return Channels.newOutputStream(channel); - } - - @Override - public void close() throws IOException { - - channel.close(); - } - } - protected final FileSystem fileSystem; /** @@ -192,7 +86,7 @@ public ReadData createReadData(final String normalPath) { public LockedFileChannel lockForReading(final String normalPath) throws N5IOException { try { - return new LockedFileChannel(normalPath, true); + return FILE_LOCK_MANAGER.lockForReading(fileSystem.getPath(normalPath)); } catch (final NoSuchFileException e) { throw new N5NoSuchKeyException("No such file", e); } catch (IOException | UncheckedIOException e) { @@ -204,7 +98,7 @@ public LockedFileChannel lockForReading(final String normalPath) throws N5IOExce public LockedChannel lockForWriting(final String normalPath) throws N5IOException { try { - return new LockedFileChannel(normalPath, false); + return FILE_LOCK_MANAGER.lockForWriting(fileSystem.getPath(normalPath)); } catch (final NoSuchFileException e) { throw new N5NoSuchKeyException("No such file", e); } catch (IOException | UncheckedIOException e) { @@ -212,10 +106,10 @@ public LockedChannel lockForWriting(final String normalPath) throws N5IOExceptio } } - public LockedChannel lockForReading(final Path path) throws N5IOException { + public LockedFileChannel lockForReading(final Path path) throws N5IOException { try { - return new LockedFileChannel(path, true); + return FILE_LOCK_MANAGER.lockForReading(path); } catch (final NoSuchFileException e) { throw new N5NoSuchKeyException("No such file", e); } catch (IOException | UncheckedIOException e) { @@ -223,10 +117,10 @@ public LockedChannel lockForReading(final Path path) throws N5IOException { } } - public LockedChannel lockForWriting(final Path path) throws N5IOException { + public LockedFileChannel lockForWriting(final Path path) throws N5IOException { try { - return new LockedFileChannel(path, false); + return FILE_LOCK_MANAGER.lockForWriting(path); } catch (final NoSuchFileException e) { throw new N5NoSuchKeyException("No such file", e); } catch (IOException | UncheckedIOException e) { @@ -611,7 +505,7 @@ public long size() { @Override public ReadData materialize(final long offset, final long length) { - try (final LockedFileChannel lfs = new LockedFileChannel(normalKey, true)) { + try (final LockedFileChannel lfs = lockForReading(fileSystem.getPath(normalKey))) { final FileChannel channel = lfs.getFileChannel(); channel.position(offset); if (length > Integer.MAX_VALUE) diff --git a/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java b/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java new file mode 100644 index 000000000..b002d6443 --- /dev/null +++ b/src/main/java/org/janelia/saalfeldlab/n5/KeyLockState.java @@ -0,0 +1,209 @@ +package org.janelia.saalfeldlab.n5; + +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.concurrent.Semaphore; + +/** + * Per-key state that tracks both thread locks and file locks. + */ +class KeyLockState { + + private final Path path; + + public KeyLockState(final Path path) { + this.path = path; + } + + /** + * The current system-level file lock (shared for writing, non-shared for + * reading). Is {@link ChannelLock#close closed} when the last {@link + * #releaseRead() Reader is released}, or the (one and only) {@link + * #releaseWrite() Writer is released}. + */ + private volatile ChannelLock channelLock; + + /** + * Multiple Readers coordinate via this mutex. {@code numReaders} may only + * be modified when {@code readerMutex} is held. + */ + private final Semaphore readerMutex = new Semaphore(1); + private int numReaders = 0; + + /** + * This coordinates mutual exclusion between one writer and (the first of) + * multiple readers. {@code channelLock} may only be created or closed when + * {@code channelLockMutex} is held. + */ + private final Semaphore channelLockMutex = new Semaphore(1); + + LockedFileChannel acquireRead() throws IOException { + + try { + readerMutex.acquire(); + try { + if (numReaders == 0) { + // We are the first Reader, and are responsible for creating the channelLock + // (Other concurrent Readers will still be blocked in readerMutex.) + + // If a Writer is still open, this will block us until the Writer is closed. + channelLockMutex.acquire(); + + try { + channelLock = ChannelLock.lock(path, false); + } catch (IOException e) { + // Something went wrong. Back off. + channelLockMutex.release(); + throw e; + } + } + + // 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) { + releaseChannelLock(); + } + throw e; + } + + // We have a FileChannel. + // Create a LockedFileChannel that will releaseRead() when it is closed. + ++numReaders; + return new LockedFileChannel(channel, this::releaseRead); + } finally { + readerMutex.release(); + } + } catch (InterruptedException e) { + throw new IOException(e); + } + } + + 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 { + readerMutex.acquire(); + try { + --numReaders; + if (numReaders == 0) { + // We were the last Reader, and are responsible for releasing the channelLock + releaseChannelLock(); + } + } finally { + readerMutex.release(); + } + } catch (InterruptedException e) { + throw new IOException(e); + } + } + + private void releaseChannelLock() throws IOException { + + try { + channelLock.close(); + } finally { + channelLockMutex.release(); + } + } + + LockedFileChannel acquireWrite() throws IOException { + + try { + // If another Writer or Reader is still open, this will block until it is closed. + channelLockMutex.acquire(); + + try { + channelLock = ChannelLock.lock(path, true); + } catch (IOException e) { + // Something went wrong. Back off. + channelLockMutex.release(); + throw e; + } + + // We have a WRITE ChannelLock. + // Create a LockedFileChannel that will releaseWrite() when it is closed. + return new LockedFileChannel(channelLock.getChannel(), this::releaseWrite); + } catch (InterruptedException e) { + throw new IOException(e); + } + } + + 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/main/java/org/janelia/saalfeldlab/n5/LockedFileChannel.java b/src/main/java/org/janelia/saalfeldlab/n5/LockedFileChannel.java new file mode 100644 index 000000000..b678dfe41 --- /dev/null +++ b/src/main/java/org/janelia/saalfeldlab/n5/LockedFileChannel.java @@ -0,0 +1,86 @@ +package org.janelia.saalfeldlab.n5; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.Reader; +import java.io.Writer; +import java.nio.channels.Channels; +import java.nio.channels.FileChannel; +import java.nio.charset.StandardCharsets; + +/** + * LockedFileChannel implementation for both read and write operations. + */ +// TODO: This only has to be public because of a test in another package. Fix that +public class LockedFileChannel implements LockedChannel { + + private final FileChannel channel; + private ReleaseLock releaseLock; + + @FunctionalInterface + public interface ReleaseLock { + + void release() throws IOException; + } + + LockedFileChannel(final FileChannel channel, final ReleaseLock releaseLock) { + + this.channel = channel; + this.releaseLock = releaseLock; + } + + @Override + public Reader newReader() throws N5Exception.N5IOException { + + return Channels.newReader(channel, StandardCharsets.UTF_8.name()); + } + + @Override + public InputStream newInputStream() throws N5Exception.N5IOException { + + return Channels.newInputStream(channel); + } + + @Override + public Writer newWriter() throws N5Exception.N5IOException { + + truncateChannel(); + return Channels.newWriter(channel, StandardCharsets.UTF_8.name()); + } + + @Override + public OutputStream newOutputStream() throws N5Exception.N5IOException { + + truncateChannel(); + return Channels.newOutputStream(channel); + } + + // TODO: This only has to be public because of a test in another package. Fix that + public FileChannel getFileChannel() { + + return channel; + } + + private void truncateChannel() throws N5Exception.N5IOException { + + try { + channel.truncate(0); + } catch (final IOException e) { + throw new N5Exception.N5IOException("Failed to truncate channel", e); + } + } + + @Override + public void close() throws IOException { + + channel.close(); + if (releaseLock != null) { + releaseLock.release(); + releaseLock = null; + // Mote that setting releaseLock=null here drops the (method) + // reference to LockKeyState, which potentially allows clearing the + // WeakReference that FileLockManager holds. + } + } +} diff --git a/src/test/java/org/janelia/saalfeldlab/n5/FileKeyLockManagerTest.java b/src/test/java/org/janelia/saalfeldlab/n5/FileKeyLockManagerTest.java new file mode 100644 index 000000000..49cf70cca --- /dev/null +++ b/src/test/java/org/janelia/saalfeldlab/n5/FileKeyLockManagerTest.java @@ -0,0 +1,448 @@ +/*- + * #%L + * Not HDF5 + * %% + * Copyright (C) 2017 - 2025 Stephan Saalfeld + * %% + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright notice, + * this list of conditions and the following disclaimer in the documentation + * and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + * #L% + */ +package org.janelia.saalfeldlab.n5; + +import static org.janelia.saalfeldlab.n5.FileKeyLockManager.FILE_LOCK_MANAGER; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.io.Closeable; +import java.io.IOException; +import java.io.Reader; +import java.io.Writer; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class FileKeyLockManagerTest { + private Path tempDir; + + @Before + public void setUp() throws IOException { + + tempDir = Files.createTempDirectory("fklm-test"); + } + + @After + public void tearDown() throws IOException { + + if (tempDir != null) { + Files.walk(tempDir) + .sorted((a, b) -> -a.compareTo(b)) + .forEach(p -> { + try { + Files.deleteIfExists(p); + } catch (IOException e) { + /* ignore */ + } + }); + } + } + + @Test + public void testConcurrentReads() throws Exception { + + /* create a test file */ + final Path testFile = tempDir.resolve("test.txt"); + final String testContent = "test content for concurrent reads"; + Files.write(testFile, testContent.getBytes()); + + final int numReaders = 5; + final ExecutorService executor = Executors.newFixedThreadPool(numReaders); + + final CountDownLatch startLatch = new CountDownLatch(1); + final CountDownLatch readersReady = new CountDownLatch(numReaders); + final CountDownLatch readersFinished = new CountDownLatch(numReaders); + final AtomicInteger concurrentReaders = new AtomicInteger(0); + final AtomicInteger maxConcurrentReaders = new AtomicInteger(0); + final AtomicInteger successfulReads = new AtomicInteger(0); + + for (int i = 0; i < numReaders; i++) { + executor.submit(() -> { + try { + readersReady.countDown(); + startLatch.await(); + + try (final LockedChannel lock = FILE_LOCK_MANAGER.lockForReading(testFile)) { + final int concurrent = concurrentReaders.incrementAndGet(); + maxConcurrentReaders.updateAndGet(max -> Math.max(max, concurrent)); + + /* actually read from the channel */ + try (final Reader reader = lock.newReader()) { + final char[] buf = new char[testContent.length()]; + final int charsRead = reader.read(buf); + if (charsRead > 0 && new String(buf, 0, charsRead).equals(testContent)) { + successfulReads.incrementAndGet(); + } + } + Thread.sleep(100); + concurrentReaders.decrementAndGet(); + } + } catch (Exception e) { + e.printStackTrace(); + } finally { + readersFinished.countDown(); + } + }); + } + + assertTrue("All readers should be ready", readersReady.await(5, TimeUnit.SECONDS)); + + final long startTime = System.currentTimeMillis(); + startLatch.countDown(); + + assertTrue("All readers should finish", readersFinished.await(10, TimeUnit.SECONDS)); + final long duration = System.currentTimeMillis() - startTime; + + executor.shutdown(); + assertTrue("Executor should terminate", executor.awaitTermination(5, TimeUnit.SECONDS)); + + System.out.println("Test completed in " + duration + "ms"); + System.out.println("Maximum concurrent readers: " + maxConcurrentReaders.get()); + System.out.println("Successful reads: " + successfulReads.get()); + + assertEquals("All readers should have read successfully", numReaders, successfulReads.get()); + assertTrue("Multiple readers should have been reading concurrently", maxConcurrentReaders.get() > 1); + assertTrue("Concurrent execution should be faster than sequential", duration < numReaders * 100); + } + + @Test + public void testReadWriteExclusion() throws Exception { + + /* create a test file */ + final Path testFile = tempDir.resolve("test2.txt"); + Files.write(testFile, "initial content".getBytes()); + + final String writtenContent = "written by writer"; + + /* acquire a write lock and write to the file */ + final LockedChannel writeLock = FILE_LOCK_MANAGER.lockForWriting(testFile); + try (final Writer writer = writeLock.newWriter()) { + writer.write(writtenContent); + } + + /* try to acquire a read lock from another thread - should block */ + final CountDownLatch readAttempted = new CountDownLatch(1); + final CountDownLatch readAcquired = new CountDownLatch(1); + final AtomicReference readContent = new AtomicReference<>(); + + new Thread(() -> { + readAttempted.countDown(); + try (final LockedChannel readLock = FILE_LOCK_MANAGER.lockForReading(testFile)) { + /* actually read from the channel */ + try (final Reader reader = readLock.newReader()) { + final char[] buf = new char[writtenContent.length()]; + final int charsRead = reader.read(buf); + if (charsRead > 0) { + readContent.set(new String(buf, 0, charsRead)); + } + } + readAcquired.countDown(); + } catch (IOException e) { + e.printStackTrace(); + } + }).start(); + + assertTrue(readAttempted.await(1, TimeUnit.SECONDS)); + + assertFalse("Read lock should not be acquired while write lock is held", + readAcquired.await(200, TimeUnit.MILLISECONDS)); + + writeLock.close(); + + assertTrue("Read lock should be acquired after write lock is released", + readAcquired.await(2, TimeUnit.SECONDS)); + + 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; + private final LockedFileChannel channel; + + CleanUpHelper() throws IOException, InterruptedException { + path = tempDir.resolve("trigger.txt"); + Files.write(path, "trigger".getBytes()); + channel = FILE_LOCK_MANAGER.lockForReading(path); + tryWaitForSize(-1, 10); + } + + private void tryWaitForSize(final int expectedSize) throws IOException, InterruptedException { + tryWaitForSize(expectedSize, 100); + } + + private void tryWaitForSize(final int expectedSize, final int numIterations) throws IOException, InterruptedException { + + // Wait a bit, trigger GC, loop a few times to try to trigger stale WeakReferences to be processed + for (int i = 0; i < numIterations; ++i) { + Thread.sleep(10); + System.gc(); + + // FileKeyLockManager.cleanUp() is called on the side during + // normal usage. We keep locking and unlocking "trigger.txt" for + // which we already hold a read lock. This will keep the + // FileKeyLockManager working, but will not lead to + // removal/insertion of KeyLockState for "trigger.txt". + try (LockedFileChannel temp = FILE_LOCK_MANAGER.lockForReading(path)) { + } + + if (FILE_LOCK_MANAGER.size() == expectedSize) { + break; + } + } + } + + @Override + public void close() throws IOException { + channel.close(); + } + } + + @Test + // TODO: Remove? This test relies on garbage collection behaviour and is inherently fragile. + public void testLockCleanup() throws Exception { + + /* create test files */ + final Path testFile1 = tempDir.resolve("key1.txt"); + final Path testFile2 = tempDir.resolve("key2.txt"); + final Path testFile3 = tempDir.resolve("key3.txt"); + final String content = "content"; + Files.write(testFile1, content.getBytes()); + Files.write(testFile2, content.getBytes()); + Files.write(testFile3, content.getBytes()); + + final CleanUpHelper cleanup = new CleanUpHelper(); + final int initialSize = FILE_LOCK_MANAGER.size(); + + final LockedChannel lock1 = FILE_LOCK_MANAGER.lockForReading(testFile1); + final LockedChannel lock2 = FILE_LOCK_MANAGER.lockForWriting(testFile2); + final LockedChannel lock3 = FILE_LOCK_MANAGER.lockForReading(testFile3); + + /* actually perform I/O on each lock */ + try (final Reader reader = lock1.newReader()) { + final char[] buf = new char[content.length()]; + reader.read(buf); + } + try (final Writer writer = lock2.newWriter()) { + writer.write("new content"); + } + try (final Reader reader = lock3.newReader()) { + final char[] buf = new char[content.length()]; + reader.read(buf); + } + + assertEquals("Should have 3 new keys", initialSize + 3, FILE_LOCK_MANAGER.size()); + + /* close lock1 - entry should be auto-removed */ + lock1.close(); + cleanup.tryWaitForSize(initialSize + 2); + assertEquals("key1 should be auto-removed on close", initialSize + 2, FILE_LOCK_MANAGER.size()); + + /* close remaining locks - entries should be auto-removed */ + lock2.close(); + lock3.close(); + cleanup.tryWaitForSize(initialSize); + assertEquals("All entries should be auto-removed on close", initialSize, FILE_LOCK_MANAGER.size()); + } + + @Test + public void testWriteLockCreatesFile() throws Exception { + + /* file does not exist - write lock creates it via CREATE option */ + final Path testFile = tempDir.resolve("newfile.txt"); + final String content = "written to new file"; + + assertFalse("File should not exist initially", Files.exists(testFile)); + + try (final LockedChannel writeLock = FILE_LOCK_MANAGER.lockForWriting(testFile)) { + assertTrue("File should be created by write lock", Files.exists(testFile)); + /* actually write to the file */ + try (final Writer writer = writeLock.newWriter()) { + writer.write(content); + } + } + + /* verify written content */ + assertEquals("Content should be written", content, new String(Files.readAllBytes(testFile))); + } + + @Test + public void testWriteLockCreatesParentDirectories() throws Exception { + + /* parent directories do not exist - write lock creates them */ + final Path testFile = tempDir.resolve("a/b/c/newfile.txt"); + final String content = "written to nested file"; + + assertFalse("File should not exist initially", Files.exists(testFile)); + assertFalse("Parent should not exist initially", Files.exists(testFile.getParent())); + + try (final LockedChannel writeLock = FILE_LOCK_MANAGER.lockForWriting(testFile)) { + assertTrue("File should be created by write lock", Files.exists(testFile)); + assertTrue("Parent directories should be created", Files.exists(testFile.getParent())); + /* actually write to the file */ + try (final Writer writer = writeLock.newWriter()) { + writer.write(content); + } + } + + /* verify written content */ + assertEquals("Content should be written", content, new String(Files.readAllBytes(testFile))); + } + + @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)); + } + + @Test + // TODO: Remove? This test relies on garbage collection behaviour and is inherently fragile. + public void testLocksMapEmptyAfterProperClose() throws Exception { + + final Path testFile = tempDir.resolve("proper-close.txt"); + Files.write(testFile, "content".getBytes()); + + final CleanUpHelper cleanup = new CleanUpHelper(); + final int initialSize = FILE_LOCK_MANAGER.size(); + + try (final LockedChannel lock = FILE_LOCK_MANAGER.lockForWriting(testFile); + final Writer writer = lock.newWriter()) { + writer.write("new content"); + } + + cleanup.tryWaitForSize(initialSize); + assertEquals("locks map should be back to initial size after proper close", + initialSize, FILE_LOCK_MANAGER.size()); + } + + + @Test + // TODO: Remove? This test relies on garbage collection behaviour and is inherently fragile. + public void testLocksMapEmptyAfterLeakedChannelIsGCd() throws Exception { + + final Path testFile = tempDir.resolve("leaked-channel.txt"); + Files.write(testFile, "content".getBytes()); + + final CleanUpHelper cleanup = new CleanUpHelper(); + final int initialSize = FILE_LOCK_MANAGER.size(); + + /* acquire lock in a separate scope so reference can be GC'd */ + acquireAndLeakLock(testFile); + + cleanup.tryWaitForSize(initialSize); + + assertEquals("locks map should be back to initial size after leaked channel is GC'd", + initialSize, FILE_LOCK_MANAGER.size()); + } + + private void acquireAndLeakLock(final Path path) throws IOException { + /* acquire lock but don't close it - just let the reference go out of scope */ + + @SuppressWarnings("resource") + LockedChannel leaked = FILE_LOCK_MANAGER.lockForWriting(path); + try (final Writer writer = leaked.newWriter()) { + writer.write("leaked content"); + } + } + +} diff --git a/src/test/java/org/janelia/saalfeldlab/n5/N5FSTest.java b/src/test/java/org/janelia/saalfeldlab/n5/N5FSTest.java index 31339af06..64bd88a3c 100644 --- a/src/test/java/org/janelia/saalfeldlab/n5/N5FSTest.java +++ b/src/test/java/org/janelia/saalfeldlab/n5/N5FSTest.java @@ -28,6 +28,9 @@ */ package org.janelia.saalfeldlab.n5; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import java.io.File; @@ -38,13 +41,17 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.attribute.FileAttribute; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Ignore; import org.junit.Test; import com.google.gson.GsonBuilder; @@ -79,7 +86,8 @@ protected String tempN5Location() throws URISyntaxException { return new URI("file", null, basePath, null).toString(); } - @Override protected N5Writer createN5Writer() throws IOException, URISyntaxException { + @Override + protected N5Writer createN5Writer() throws IOException, URISyntaxException { return createN5Writer(tempN5Location(), new GsonBuilder()); } @@ -163,49 +171,109 @@ public void testWriteLock() throws IOException { exec.shutdownNow(); } - + @Test - public void testLockReleaseByReader() throws IOException, ExecutionException, InterruptedException, TimeoutException { + public void testFSLockRelease() throws IOException, ExecutionException, InterruptedException, TimeoutException { final Path path = Paths.get(tempN5PathName(), "lock"); - final LockedChannel lock = access.lockForWriting(path); - - lock.newReader().close(); + final ExecutorService exec = Executors.newFixedThreadPool(2); + + // first thread acquires the lock, waits for 200ms then should release it + exec.submit(() -> { + try( final LockedChannel lock = access.lockForWriting(path)) { + lock.newReader(); + Thread.sleep(200); + } catch (IOException e) { + e.printStackTrace(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + }); - final ExecutorService exec = Executors.newSingleThreadExecutor(); + // second thread waits for the lock. + // it should get it within a few seconds. final Future future = exec.submit(() -> { access.lockForWriting(path).close(); return null; }); future.get(3, TimeUnit.SECONDS); - future.cancel(true); - lock.close(); Files.delete(path); + exec.shutdownNow(); + } + + @Test + public void testReadLockBehavior() throws IOException, InterruptedException, ExecutionException, TimeoutException { + + final Path path = Paths.get(tempN5PathName(), "read-lock"); + path.toFile().createNewFile(); + + final ExecutorService exec = Executors.newFixedThreadPool(3); + + final AtomicBoolean v = new AtomicBoolean(false); + + // first thread acquires a read lock, waits for 200ms + Future f = exec.submit(() -> { + try( final LockedChannel lock = access.lockForReading(path)) { + lock.newReader(); + Thread.sleep(200); + + // ensure that the other thread updated the value + assertTrue(v.get()); + } catch (IOException e) { + e.printStackTrace(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + }); + + // second thread gets a read lock + // and should not be blocked + // this thread updates the boolean + exec.submit(() -> { + try( final LockedChannel lock = access.lockForReading(path)) { + lock.newReader(); + v.set(true); + } + return null; + }); + + f.get(3, TimeUnit.SECONDS); exec.shutdownNow(); + Files.delete(path); } @Test - public void testLockReleaseByInputStream() throws IOException, ExecutionException, InterruptedException, TimeoutException { + public void testWriteLockBehavior() throws IOException, ExecutionException, InterruptedException, TimeoutException { - final Path path = Paths.get(tempN5PathName(), "lock"); - final LockedChannel lock = access.lockForWriting(path); - lock.newInputStream().close(); + final Path path = Paths.get(tempN5PathName(), "lock"); + final ExecutorService exec = Executors.newFixedThreadPool(2); + + // first thread acquires the lock, waits for 200ms then should release it + exec.submit(() -> { + try( final LockedChannel lock = access.lockForWriting(path)) { + lock.newReader(); + Thread.sleep(200); + } catch (IOException e) { + e.printStackTrace(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + }); - final ExecutorService exec = Executors.newSingleThreadExecutor(); + // second thread waits for the lock. + // it should get it within a few seconds. final Future future = exec.submit(() -> { access.lockForWriting(path).close(); return null; }); future.get(3, TimeUnit.SECONDS); - future.cancel(true); - lock.close(); Files.delete(path); - exec.shutdownNow(); } + } diff --git a/src/test/java/org/janelia/saalfeldlab/n5/shard/ShardTest.java b/src/test/java/org/janelia/saalfeldlab/n5/shard/ShardTest.java index 5d5af2715..2b1cc4e95 100644 --- a/src/test/java/org/janelia/saalfeldlab/n5/shard/ShardTest.java +++ b/src/test/java/org/janelia/saalfeldlab/n5/shard/ShardTest.java @@ -28,21 +28,12 @@ */ package org.janelia.saalfeldlab.n5.shard; -import org.janelia.saalfeldlab.n5.ByteArrayDataBlock; -import org.janelia.saalfeldlab.n5.DataBlock; -import org.janelia.saalfeldlab.n5.DataType; -import org.janelia.saalfeldlab.n5.DatasetAttributes; -import org.janelia.saalfeldlab.n5.FileSystemKeyValueAccess; -import org.janelia.saalfeldlab.n5.KeyValueAccess; -import org.janelia.saalfeldlab.n5.KeyValueAccessReadData; -import org.janelia.saalfeldlab.n5.N5Exception; -import org.janelia.saalfeldlab.n5.N5FSTest; -import org.janelia.saalfeldlab.n5.N5KeyValueWriter; -import org.janelia.saalfeldlab.n5.N5Writer; -import org.janelia.saalfeldlab.n5.RawCompression; +import com.google.gson.GsonBuilder; +import org.janelia.saalfeldlab.n5.*; import org.janelia.saalfeldlab.n5.N5Exception.N5NoSuchKeyException; -import org.janelia.saalfeldlab.n5.GsonKeyValueN5Writer; -import org.janelia.saalfeldlab.n5.codec.*; +import org.janelia.saalfeldlab.n5.codec.DataCodecInfo; +import org.janelia.saalfeldlab.n5.codec.N5BlockCodecInfo; +import org.janelia.saalfeldlab.n5.codec.RawBlockCodecInfo; import org.janelia.saalfeldlab.n5.readdata.ReadData; import org.janelia.saalfeldlab.n5.shard.ShardIndex.IndexLocation; import org.junit.After; @@ -51,11 +42,6 @@ import org.junit.runner.RunWith; import org.junit.runners.Parameterized; -import com.google.gson.GsonBuilder; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - import java.io.File; import java.io.IOException; import java.io.UncheckedIOException; @@ -68,15 +54,12 @@ import java.nio.file.FileSystems; import java.nio.file.Files; import java.nio.file.NoSuchFileException; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.function.Function; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + @RunWith(Parameterized.class) public class ShardTest { @@ -627,8 +610,8 @@ public ReadData materialize(final long offset, final long length) { numMaterializeCalls++; - try (final TrackingLockedFileChannel lfs = new TrackingLockedFileChannel(normalKey, true); - final FileChannel channel = lfs.getFileChannel()) { + try (final LockedFileChannel lfs = lockForReading(normalKey)) { + final FileChannel channel = lfs.getFileChannel(); channel.position(offset); if (length > Integer.MAX_VALUE) @@ -652,19 +635,6 @@ public ReadData materialize(final long offset, final long length) { } } } - - private class TrackingLockedFileChannel extends LockedFileChannel { - - protected TrackingLockedFileChannel(String path, boolean readOnly) throws IOException { - super(path, readOnly); - } - - @Override - protected FileChannel getFileChannel() { - return channel; - } - } - private static boolean validBounds(long channelSize, long offset, long length) { if (offset < 0)