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 d40cbe31ad78fe09372e47721d5074359d6c7744..ad55206fe7afd25a58e9351cf88e30842ac84863 100644 (file)
--- 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 @@ public abstract class AbstractWriteBuffer implements WriteBuffer {
                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 abstract class AbstractWriteBuffer implements WriteBuffer {
                                }
                        }
 
-                       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 abstract class AbstractWriteBuffer implements WriteBuffer {
                        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 @@ public abstract class AbstractWriteBuffer implements WriteBuffer {
                        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;
        }