DHT: Fix thread-safety issue in AbstractWriteBuffer There is a data corruption issue with the 'running' list if a background thread schedules something onto the buffer while the application thread is also using it. Change-Id: I5ba78b98b6632965d677a9c8f209f0cf8320cc3d Signed-off-by: Shawn O. Pearce <spearce@spearce.org>
diff --git a/org.eclipse.jgit.storage.dht/src/org/eclipse/jgit/storage/dht/spi/util/AbstractWriteBuffer.java b/org.eclipse.jgit.storage.dht/src/org/eclipse/jgit/storage/dht/spi/util/AbstractWriteBuffer.java index d40cbe3..ad55206 100644 --- a/org.eclipse.jgit.storage.dht/src/org/eclipse/jgit/storage/dht/spi/util/AbstractWriteBuffer.java +++ b/org.eclipse.jgit.storage.dht/src/org/eclipse/jgit/storage/dht/spi/util/AbstractWriteBuffer.java
@@ -82,6 +82,8 @@ public abstract class AbstractWriteBuffer implements WriteBuffer { private final List<Future<?>> running; + private final Object runningLock; + private final Semaphore spaceAvailable; private int queuedCount; @@ -102,6 +104,7 @@ protected AbstractWriteBuffer(ExecutorService executor, int bufferSize) { this.executor = executor; this.bufferSize = bufferSize; this.running = new LinkedList<Future<?>>(); + this.runningLock = new Object(); this.spaceAvailable = new Semaphore(bufferSize); } @@ -189,14 +192,18 @@ public void flush() throws DhtException { } } - checkRunningTasks(true); + synchronized (runningLock) { + checkRunningTasks(true); + } } finally { flushing = false; } } public void abort() throws DhtException { - checkRunningTasks(true); + synchronized (runningLock) { + checkRunningTasks(true); + } } private void acquireSpace(int sz) throws DhtException { @@ -259,9 +266,11 @@ public T call() throws Exception { return; } - if (!flushing) - checkRunningTasks(false); - running.add(executor.submit(op)); + synchronized (runningLock) { + if (!flushing) + checkRunningTasks(false); + running.add(executor.submit(op)); + } } /** @@ -284,8 +293,10 @@ protected <T> AsyncCallback<T> wrap(final AsyncCallback<T> callback, int size) throws DhtException { int permits = permitsForSize(size); WrappedCallback<T> op = new WrappedCallback<T>(callback, permits); - checkRunningTasks(false); - running.add(op); + synchronized (runningLock) { + checkRunningTasks(false); + running.add(op); + } return op; }