Skip to content

Commit be3fafb

Browse files
committed
fix: toward correct locking mechanism
* multiple read locks can be held across threads * only a single write lock can be held * still to do: holding a read lock does not prevent writing see #141
1 parent 3955cdf commit be3fafb

5 files changed

Lines changed: 522 additions & 35 deletions

File tree

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

Lines changed: 42 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,6 @@
6868
import java.nio.ByteBuffer;
6969
import java.nio.channels.Channels;
7070
import java.nio.channels.FileChannel;
71-
import java.nio.channels.OverlappingFileLockException;
7271
import java.nio.charset.StandardCharsets;
7372
import java.nio.file.DirectoryNotEmptyException;
7473
import java.nio.file.FileAlreadyExistsException;
@@ -84,6 +83,7 @@
8483
import java.util.Arrays;
8584
import java.util.Comparator;
8685
import java.util.Iterator;
86+
import java.util.concurrent.locks.Lock;
8787
import java.util.stream.Stream;
8888

8989
import org.janelia.saalfeldlab.n5.readdata.ReadData;
@@ -103,21 +103,51 @@ public class FileSystemKeyValueAccess implements KeyValueAccess {
103103
* {@link FileSystem} supports that. If the {@link FileSystem} does not
104104
* support locking, it returns immediately.
105105
*/
106-
protected class LockedFileChannel implements LockedChannel {
106+
protected class LockedFileChannel implements LockedChannel, AutoCloseable {
107107

108108
protected final FileChannel channel;
109109

110+
protected Lock lock;
111+
110112
protected LockedFileChannel(final String path, final boolean readOnly) throws IOException {
111113

112114
this(fileSystem.getPath(path), readOnly);
113115
}
114116

115117
protected LockedFileChannel(final Path path, final boolean readOnly) throws IOException {
116118

119+
final String key = path.toAbsolutePath().toString();
120+
boolean haveLock = false;
121+
while( !haveLock ) {
122+
123+
if (readOnly) {
124+
lock = keyLock.tryLockForReading(key);
125+
}
126+
else {
127+
lock = keyLock.tryLockForWriting(key);
128+
}
129+
130+
if (lock != null) {
131+
haveLock = true;
132+
break;
133+
}
134+
135+
try {
136+
Thread.sleep(100);
137+
} catch (InterruptedException e) {
138+
throw new N5Exception(e);
139+
}
140+
}
141+
117142
final OpenOption[] options;
118143
if (readOnly) {
119144
options = new OpenOption[]{StandardOpenOption.READ};
120-
channel = FileChannel.open(path, options);
145+
try {
146+
channel = FileChannel.open(path, options);
147+
} catch (final NoSuchFileException e) {
148+
lock.unlock();
149+
throw e;
150+
}
121151
} else {
122152
options = new OpenOption[]{StandardOpenOption.READ, StandardOpenOption.WRITE,
123153
StandardOpenOption.CREATE};
@@ -132,20 +162,6 @@ protected LockedFileChannel(final Path path, final boolean readOnly) throws IOEx
132162
}
133163
}
134164

135-
for (boolean waiting = true; waiting;) {
136-
waiting = false;
137-
try {
138-
channel.lock(0L, Long.MAX_VALUE, readOnly);
139-
} catch (final OverlappingFileLockException e) {
140-
waiting = true;
141-
try {
142-
Thread.sleep(100);
143-
} catch (final InterruptedException f) {
144-
waiting = false;
145-
Thread.currentThread().interrupt();
146-
}
147-
} catch (final IOException e) {}
148-
}
149165
}
150166

151167
protected FileChannel getFileChannel() {
@@ -191,11 +207,13 @@ public OutputStream newOutputStream() throws N5IOException {
191207

192208
@Override
193209
public void close() throws IOException {
194-
210+
lock.unlock();
195211
channel.close();
196212
}
197213
}
198214

215+
protected final KeyLock keyLock;
216+
199217
protected final FileSystem fileSystem;
200218

201219
/**
@@ -206,6 +224,12 @@ public void close() throws IOException {
206224
public FileSystemKeyValueAccess(final FileSystem fileSystem) {
207225

208226
this.fileSystem = fileSystem;
227+
keyLock = new KeyLock();
228+
}
229+
230+
public KeyLock getLocks() {
231+
232+
return keyLock;
209233
}
210234

211235
@Override
Lines changed: 164 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,164 @@
1+
/*-
2+
* #%L
3+
* Not HDF5
4+
* %%
5+
* Copyright (C) 2017 - 2025 Stephan Saalfeld
6+
* %%
7+
* Redistribution and use in source and binary forms, with or without
8+
* modification, are permitted provided that the following conditions are met:
9+
*
10+
* 1. Redistributions of source code must retain the above copyright notice,
11+
* this list of conditions and the following disclaimer.
12+
* 2. Redistributions in binary form must reproduce the above copyright notice,
13+
* this list of conditions and the following disclaimer in the documentation
14+
* and/or other materials provided with the distribution.
15+
*
16+
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
17+
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
18+
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
19+
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR CONTRIBUTORS BE
20+
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
21+
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
22+
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
23+
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
24+
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
25+
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
26+
* POSSIBILITY OF SUCH DAMAGE.
27+
* #L%
28+
*/
29+
package org.janelia.saalfeldlab.n5;
30+
31+
import java.util.Optional;
32+
import java.util.concurrent.ConcurrentHashMap;
33+
import java.util.concurrent.locks.ReentrantReadWriteLock;
34+
import java.util.concurrent.locks.Lock;
35+
36+
/**
37+
* A lock manager that provides thread-safe read/write locking for keys.
38+
*
39+
* This class manages a set of {@link ReentrantReadWriteLock}s, one per key,
40+
* allowing multiple threads to read the same key simultaneously while ensuring
41+
* exclusive access for writes.
42+
*
43+
* Unlike file locks which operate at the process level, this provides
44+
* thread-level locking within a single JVM.
45+
*/
46+
public class KeyLock {
47+
48+
private final ConcurrentHashMap<String, ReentrantReadWriteLock> locks = new ConcurrentHashMap<>();
49+
50+
/**
51+
* Acquires a read lock for the specified key. Multiple threads can hold
52+
* read locks for the same key simultaneously.
53+
*
54+
* @param key
55+
* the key to lock for reading
56+
* @return a {@link Lock} that must be unlocked when done
57+
*/
58+
public Lock lockForReading(String key) {
59+
60+
ReentrantReadWriteLock rwLock = locks.computeIfAbsent(key, k -> new ReentrantReadWriteLock());
61+
Lock readLock = rwLock.readLock();
62+
readLock.lock();
63+
return readLock;
64+
}
65+
66+
/**
67+
* Acquires a write lock for the specified key. Only one thread can hold a
68+
* write lock for a key at a time.
69+
*
70+
* @param key
71+
* the key to lock for writing
72+
* @return a {@link Lock} that must be unlocked when done
73+
*/
74+
public Lock lockForWriting(String key) {
75+
76+
ReentrantReadWriteLock rwLock = locks.computeIfAbsent(key, k -> new ReentrantReadWriteLock());
77+
Lock writeLock = rwLock.writeLock();
78+
writeLock.lock();
79+
return writeLock;
80+
}
81+
82+
/**
83+
* Attempts to acquire a read lock for the specified key without blocking.
84+
*
85+
* @param key
86+
* the key to lock for reading
87+
* @return a {@link Lock} if the lock was acquired, null otherwise
88+
*/
89+
public Lock tryLockForReading(String key) {
90+
91+
ReentrantReadWriteLock rwLock = locks.computeIfAbsent(key, k -> new ReentrantReadWriteLock());
92+
Lock readLock = rwLock.readLock();
93+
if (readLock.tryLock()) {
94+
return readLock;
95+
}
96+
return null;
97+
}
98+
99+
/**
100+
* Attempts to acquire a write lock for the specified key without blocking.
101+
*
102+
* @param key
103+
* the key to lock for writing
104+
* @return a {@link Lock} if the lock was acquired, null otherwise
105+
*/
106+
public Lock tryLockForWriting(String key) {
107+
108+
ReentrantReadWriteLock rwLock = locks.computeIfAbsent(key, k -> new ReentrantReadWriteLock());
109+
Lock writeLock = rwLock.writeLock();
110+
if (writeLock.tryLock()) {
111+
return writeLock;
112+
}
113+
return null;
114+
}
115+
116+
/**
117+
* Returns the number of keys currently being tracked.
118+
*
119+
* @return the number of keys with associated locks
120+
*/
121+
public int size() {
122+
123+
return locks.size();
124+
}
125+
126+
/**
127+
* Removes the lock for a key if it is not currently held. This can be used
128+
* to clean up unused locks to prevent memory leaks.
129+
*
130+
* @param key
131+
* the key whose lock should be removed
132+
* @return true if the lock was removed, false if it's currently in use
133+
*/
134+
public boolean removeLockIfUnused(String key) {
135+
136+
ReentrantReadWriteLock rwLock = locks.get(key);
137+
if (rwLock != null && !rwLock.isWriteLocked() && rwLock.getReadLockCount() == 0) {
138+
return locks.remove(key, rwLock);
139+
}
140+
return false;
141+
}
142+
143+
public Optional<ReentrantReadWriteLock> getKeyLock(String key) {
144+
// TODO doc me
145+
return Optional.ofNullable(locks.get(key));
146+
}
147+
148+
/**
149+
* Clears all unused locks from the lock map. Locks that are currently held
150+
* will not be removed.
151+
*
152+
* @return the number of locks that were removed
153+
*/
154+
public int clearUnusedLocks() {
155+
156+
int removed = 0;
157+
for (String key : locks.keySet()) {
158+
if (removeLockIfUnused(key)) {
159+
removed++;
160+
}
161+
}
162+
return removed;
163+
}
164+
}

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

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,6 @@
5656
import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
5757

5858
import java.io.Closeable;
59-
import java.io.IOException;
6059
import java.io.InputStream;
6160
import java.io.OutputStream;
6261
import java.io.Reader;

0 commit comments

Comments
 (0)