Revise FileKeyLockManager.

Locks are not tied to the locking thread anymore. (A LockedChannel can be
created by one thread and then be released by another thread). This required to
replace ReentrantReadWriteLock with manual reader/writer counting guarded by
Semaphores.

LockedFileChannel is now created by KeyLockState (instead of FileKeyLockManager)
because for releasing we need to distinguish between read and write locks.

I revised clean-up of stale KeyLockStates by putting WeakReferences in the
ConcurrentHashMap and adding a ReferenceQueue to remove entries whose
KeyLockState has been GCed.

I put no special logic to handle leaked LockedFileChannels (that are abandoned
without being properly closed). Surprisingly, this still works. When the
KeyLockState is GCed, its associated ChannelLock along with the existing
FileChannel and FileLock is also GCed. At least on MacOS, this causes the JVM to
release the system-level lock, and everything works out fine.

Author: tpietzsch

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

@@ -0,0 +1,124 @@
+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 {
+		lock.release();
+		channel.close();
+	}
+
+	/**
+	 * 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).
+	 * <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.
+	 *
+	 * @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.
+	 * <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.
+	 *
+	 * @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) {
+			}
+		}
+	}
+}

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

@@ -29,13 +29,9 @@
 package org.janelia.saalfeldlab.n5;
 
 import java.io.IOException;
-import java.lang.ref.PhantomReference;
-import java.lang.ref.Reference;
 import java.lang.ref.ReferenceQueue;
-import java.nio.channels.FileChannel;
+import java.lang.ref.WeakReference;
 import java.nio.file.Path;
-import java.nio.file.StandardOpenOption;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
@@ -46,70 +42,64 @@ class FileKeyLockManager {
 
 	static final FileKeyLockManager FILE_LOCK_MANAGER = new FileKeyLockManager();
 
-	private static final ReferenceQueue<LockedFileChannel> queue = new ReferenceQueue<>();
-	private static final Set<LockedChannelReference> phantomRefs = ConcurrentHashMap.newKeySet();
-	private static final ConcurrentHashMap<String, KeyLockState> locks = new ConcurrentHashMap<>();
+	private FileKeyLockManager() {
+		// singleton
+	}
 
-	/**
-	 * PhantomReference for LockedFileChannel that releases the lock if the
-	 * channel is garbage collected without being closed.
-	 */
-	private static class LockedChannelReference extends PhantomReference<LockedFileChannel> {
 
-		private final String key;
-		private final KeyLockState state;
+	private final ConcurrentHashMap<String, WeakValue> locks = new ConcurrentHashMap<>();
 
-		LockedChannelReference(final LockedFileChannel referent, final String key, final KeyLockState state) {
-			super(referent, queue);
-			this.key = key;
-			this.state = state;
-		}
+	private final ReferenceQueue<KeyLockState> refQueue = new ReferenceQueue<>();
 
-		void cleanup() {
-			state.releaseFileLockForCleanup();
-			phantomRefs.remove(this);
-			/* always remove from map - if the LockedFileChannel was GC'd, the entry is stale */
-			locks.remove(key, state);
+	private static class WeakValue extends WeakReference<KeyLockState> {
+
+		final String key;
+
+		WeakValue(
+				final String key,
+				final KeyLockState value,
+				final ReferenceQueue<KeyLockState> queue) {
+
+			super(value, queue);
+			this.key = key;
 		}
 	}
 
-	private void clearQueue() {
-		Reference<? extends LockedFileChannel> ref;
-		while ((ref = queue.poll()) != null) {
-			((LockedChannelReference)ref).cleanup();
+	/**
+	 * 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 class ManagedLockedFileChannel extends LockedFileChannel {
+	private KeyLockState keyLockState(final Path path) {
 
-		private final String key;
-		private final KeyLockState state;
+		final String key = path.toAbsolutePath().toString();
 
-		ManagedLockedFileChannel(final KeyLockState state, final FileChannel channel, final String key) {
-			super(state, channel);
-			this.key = key;
-			this.state = state;
-		}
+		cleanUp();
 
-		@Override
-		public void close() throws IOException {
-			super.close();
-			synchronized (state) {
-				if (!state.isLocked()) {
-					locks.remove(key, state);
-				}
-			}
-			clearQueue();
+		final WeakValue existingRef = locks.get(key);
+		KeyLockState state = existingRef == null ? null : existingRef.get();
+		if (state != null) {
+			return state;
 		}
-	}
 
-	private FileKeyLockManager() {
-	}
-
-	private LockedFileChannel registerLockedFileChannel(final FileChannel channel, final String key, final KeyLockState state) {
-		final LockedFileChannel lockedChannel = new ManagedLockedFileChannel(state, channel, key);
-		phantomRefs.add(new LockedChannelReference(lockedChannel, key, state));
-		return lockedChannel;
+		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;
 	}
 
 	/**
@@ -126,22 +116,8 @@ private LockedFileChannel registerLockedFileChannel(final FileChannel channel, f
 	 *             if acquiring the file lock fails
 	 */
 	public LockedFileChannel lockForReading(final Path path) throws IOException {
-		/* if there are stale entries in the queue, clean them before we try to lock */
-		clearQueue();
-
-		final String key = path.toAbsolutePath().toString();
-		final KeyLockState state = locks.computeIfAbsent(key, k -> new KeyLockState(path));
-
-		state.lockForReading();
 
-		final FileChannel channel;
-		try {
-			channel = FileChannel.open(path, StandardOpenOption.READ);
-		} catch (final IOException e) {
-			state.releaseLock();
-			throw e;
-		}
-		return registerLockedFileChannel(channel, key, state);
+		return keyLockState(path).acquireRead();
 	}
 
 	/**
@@ -155,23 +131,8 @@ public LockedFileChannel lockForReading(final Path path) throws IOException {
 	 *             if acquiring the file lock fails
 	 */
 	public LockedFileChannel lockForWriting(final Path path) throws IOException {
-		/* if there are stale entries in the queue, clean them before we try to lock */
-		clearQueue();
-
-		final String key = path.toAbsolutePath().toString();
-		final KeyLockState state = locks.computeIfAbsent(key, k -> new KeyLockState(path));
-
-		state.lockForWriting();
 
-		final FileChannel channel;
-		try {
-			channel = FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.CREATE);
-		} catch (final IOException e) {
-			state.releaseLock();
-			throw e;
-		}
-
-		return registerLockedFileChannel(channel, key, state);
+		return keyLockState(path).acquireWrite();
 	}
 
 	/**
@@ -182,24 +143,8 @@ public LockedFileChannel lockForWriting(final Path path) throws IOException {
 	 * @return a {@link LockedChannel} if the lock was acquired, null otherwise
 	 */
 	public LockedFileChannel tryLockForReading(final Path path) {
-		/* if there are stale entries in the queue, clean them before we try to lock */
-		clearQueue();
-
-		final String key = path.toAbsolutePath().toString();
-		final KeyLockState state = locks.computeIfAbsent(key, k -> new KeyLockState(path));
-
-		if (!state.tryLockForReading())
-			return null;
 
-		final FileChannel channel;
-		try {
-			channel = FileChannel.open(path, StandardOpenOption.READ);
-		} catch (final IOException e) {
-			state.releaseLock();
-			return null;
-		}
-
-		return registerLockedFileChannel(channel, key, state);
+		return keyLockState(path).tryAcquireRead();
 	}
 
 	/**
@@ -210,32 +155,16 @@ public LockedFileChannel tryLockForReading(final Path path) {
 	 * @return a {@link LockedChannel} if the lock was acquired, null otherwise
 	 */
 	public LockedFileChannel tryLockForWriting(final Path path) {
-		/* if there are stale entries in the queue, clean them before we try to lock */
-		clearQueue();
-
-		final String key = path.toAbsolutePath().toString();
-		final KeyLockState state = locks.computeIfAbsent(key, k -> new KeyLockState(path));
-
-		if (!state.tryLockForWriting())
-			return null;
 
-		final FileChannel channel;
-		try {
-			channel = FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.CREATE);
-		} catch (final IOException e) {
-			state.releaseLock();
-			return null;
-		}
-
-		return registerLockedFileChannel(channel, key, state);
+		return keyLockState(path).tryAcquireWrite();
 	}
 
 	/**
 	 * Returns the number of keys currently being tracked.
 	 *
 	 * @return the number of keys with associated locks
 	 */
-	public int size() {
+	int size() {
 
 		return locks.size();
 	}
@@ -247,15 +176,6 @@ public int size() {
 	 * @return true if removed, false if currently in use or not found
 	 */
 	public boolean removeLockIfUnused(final String key) {
-
-		final KeyLockState state = locks.get(key);
-		if (state == null)
-			return false;
-
-		synchronized (state) {
-			if (!state.isLocked())
-				return locks.remove(key, state);
-		}
-		return false;
+		throw new UnsupportedOperationException("TODO. REMOVE");
 	}
 }