File locking fixes and tests

pom.xml

@@ -148,6 +148,16 @@
 		<releaseProfiles>sign,deploy-to-scijava</releaseProfiles>
 
 		<n5-spec.version>4.0.0</n5-spec.version>
+
+		<!-- Only used for test scope -->
+		<n5-universe.version>2.4.0-alpha-9</n5-universe.version>
+		<n5-aws-s3.version>4.4.0-alpha-9</n5-aws-s3.version>
+		<n5-blosc.version>2.0.0-alpha-4</n5-blosc.version>
+		<n5-google-cloud.version>5.2.0-alpha-7</n5-google-cloud.version>
+		<n5-hdf5.version>2.3.0-alpha-7</n5-hdf5.version>
+		<n5-imglib2.version>7.1.0-alpha-8</n5-imglib2.version>
+		<n5-zarr.version>2.0.0-alpha-8</n5-zarr.version>
+		<n5-zstandard.version>2.0.0-alpha-4</n5-zstandard.version>
 	</properties>
 
 	<dependencies>
@@ -224,6 +234,13 @@
 			<scope>test</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>info.picocli</groupId>
+			<artifactId>picocli</artifactId>
+			<version>4.3.2</version>
+			<scope>test</scope>
+		</dependency>
+
 		<!-- JMH -->
 		<dependency>
 			<groupId>org.openjdk.jmh</groupId>

scripts/fsLockValidation

@@ -0,0 +1,60 @@
+#!/bin/bash
+
+#
+# Run FsLockingValidation N times in parallel.
+#
+# Usage:
+#   [N=<processes>] ./fsLockValidation [FsLockingValidation options...]
+#
+# Example:
+#   N=4 ./fsLockValidation --file /tmp/shard-test --num-repeats 5000
+
+N=${N:-2}
+
+OWN_DIR="$(readlink -f "$(dirname "${BASH_SOURCE[0]}")")"
+POM="$OWN_DIR/../pom.xml"
+TARGET="$OWN_DIR/../target"
+
+if [ ! -d "$TARGET/test-classes" ]; then
+    echo "test-classes not found -- run 'mvn test-compile' in n5/ first" >&2
+    exit 1
+fi
+
+echo "Building classpath..."
+CP=$(mvn -q -f "$POM" dependency:build-classpath \
+        -DincludeScope=test \
+        -Dmdep.outputFile=/dev/stdout \
+        2>/dev/null)
+CP="$TARGET/test-classes:$TARGET/classes:$CP"
+
+cleanup() {
+    kill "${pids[@]}" 2>/dev/null
+    wait "${pids[@]}" 2>/dev/null
+}
+trap cleanup INT TERM EXIT
+
+echo "Launching $N processes..."
+pids=()
+for i in $(seq 1 $N); do
+    java -cp "$CP" org.janelia.saalfeldlab.n5.kva.FsLockingValidation "$@" &
+    pids+=($!)
+done
+
+# Wait for all processes and collect exit codes
+all_ok=true
+for i in "${!pids[@]}"; do
+    pid=${pids[$i]}
+    wait "$pid"
+    code=$?
+    if [ $code -ne 0 ]; then
+        echo "Process $((i+1)) (pid $pid) exited with code $code" >&2
+        all_ok=false
+    fi
+done
+
+if $all_ok; then
+    echo "All $N processes completed without errors."
+else
+    echo "One or more processes reported errors." >&2
+    exit 1
+fi

scripts/writeLockTest.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+TEST_DIR=/tmp/write-lock-test.zarr
+
+[ -d "$TEST_DIR" ] && rm -r -- "$TEST_DIR"
+
+if [ ! -d "test-classes" ]; then
+    mvn test-compile
+fi
+
+mvn -e -q exec:java \
+  -Dexec.mainClass=org.janelia.saalfeldlab.n5.WriteLockExp \
+  -Dexec.classpathScope=test \
+  -Dexec.args="/tmp/write-lock-test.zarr 0" &
+pid1=$!
+
+mvn -e -q exec:java \
+  -Dexec.mainClass=org.janelia.saalfeldlab.n5.WriteLockExp \
+  -Dexec.classpathScope=test \
+  -Dexec.args="/tmp/write-lock-test.zarr 1" &
+pid2=$!
+
+trap 'kill "$pid1" "$pid2" 2>/dev/null || true' EXIT INT TERM
+
+wait "$pid1"
+wait "$pid2"

src/main/java/org/janelia/saalfeldlab/n5/ChannelLock.java

@@ -6,7 +6,9 @@
 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
@@ -17,13 +19,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.
+	 * <p>
+	 * 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 {
@@ -55,9 +66,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 LockingPolicy policy) throws IOException {
 
-		final FileChannel channel = FsIoPolicy.openFileChannel(path, forWriting);
+		final FileChannel channel = openFileChannel(path, forWriting);
+		if (policy == LockingPolicy.UNSAFE) {
+			return new ChannelLock(channel, null);
+		}
 		try {
 			while (true) {
 				try {
@@ -73,34 +87,32 @@ static ChannelLock lock(final Path path, final boolean forWriting) throws IOExce
 				}
 			}
 		} catch (Exception e) {
-			closeQuietly(channel);
-			throw e;
+			if (policy == LockingPolicy.STRICT) {
+				closeQuietly(channel);
+				throw e;
+			} else {
+				return new ChannelLock(channel, null);
+			}
 		}
 	}
 
 	/**
-	 * 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.
-	 * <p>
-	 * The {@code FileLock} is exclusive if the {@code path} is locked {@code
-	 * forWriting}, and shared otherwise.
-	 * <p>
-	 * If the {@code path} is locked {@code forWriting} non-existing file and
-	 * the parent directories are created as needed.
+	 * 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 an error occurs while opening the channel.
+	 * @throws IOException
+	 * 		if the channel cannot be opened
 	 */
-	static ChannelLock tryLock(final Path path, final boolean forWriting) throws IOException {
+	private static FileChannel openFileChannel(final Path path, final boolean forWriting) throws IOException {
 
-		FileChannel channel = null;
-		try {
-			channel = FsIoPolicy.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 (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);
 		}
 	}
 
@@ -110,6 +122,6 @@ private static void closeQuietly(final FileChannel fileChannel) {
 				fileChannel.close();
 			} catch (final IOException | UncheckedIOException ignored) {
 			}
-        }
+		}
 	}
 }

src/main/java/org/janelia/saalfeldlab/n5/FileKeyLockManager.java

@@ -31,21 +31,53 @@
 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.Collections;
+import java.util.EnumMap;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
+import static org.janelia.saalfeldlab.n5.LockingPolicy.STRICT;
+
 /**
  * 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 static final Map<LockingPolicy, FileKeyLockManager> managers = Collections.synchronizedMap(new EnumMap<>(LockingPolicy.class));
 
-	private FileKeyLockManager() {
-		// singleton
+	static FileKeyLockManager forPolicy(final LockingPolicy policy) {
+		return managers.computeIfAbsent(policy, FileKeyLockManager::new);
 	}
 
+	/**
+	 * @deprecated use {@link FileKeyLockManager#forPolicy(LockingPolicy)}
+	 */
+	@Deprecated
+	static final FileKeyLockManager FILE_LOCK_MANAGER = forPolicy(STRICT);
+
+	private final LockingPolicy policy;
+
+	/**
+	 * Create a new {@link FileKeyLockManager} with the specified locking policy.
+	 * <p>
+	 * The given locking {@link LockingPolicy 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 policy
+	 * 		the locking policy
+	 */
+	private FileKeyLockManager(final LockingPolicy policy) {
+		this.policy = policy;
+	}
 
 	private final ConcurrentHashMap<String, WeakValue> locks = new ConcurrentHashMap<>();
 
@@ -79,7 +111,7 @@ private void cleanUp()
 		}
 	}
 
-	private KeyLockState keyLockState(final Path path) {
+	private KeyLockState keyLockState(final Path path, final LockingPolicy policy) throws IOException {
 
 		final String key = path.toAbsolutePath().toString();
 
@@ -91,7 +123,7 @@ private KeyLockState keyLockState(final Path path) {
 			return state;
 		}
 
-		final KeyLockState newState = new KeyLockState(path);
+		final KeyLockState newState = new KeyLockState(path, policy);
 		while (state == null) {
 			final WeakValue ref = locks.compute(key,
 					(k, v) -> (v != null && v.get() != null)
@@ -110,53 +142,33 @@ private KeyLockState keyLockState(final Path path) {
 	 * only acquire the thread-level lock.
 	 *
 	 * @param path
-	 *            the key (file path) to lock for reading
+	 * 		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
+	 * 		if acquiring the file lock fails
 	 */
 	public LockedFileChannel lockForReading(final Path path) throws IOException {
 
-		return keyLockState(path).acquireRead();
+		return keyLockState(path, policy).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
+	 * 		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
+	 * 		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();
+		return keyLockState(path, policy).acquireWrite();
 	}
 
 	/**
@@ -168,14 +180,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");
-	}
 }