From 042a66fe8c2feef7b831b1999b8cd6e9a0181944 Mon Sep 17 00:00:00 2001
From: "Shawn O. Pearce" <spearce@spearce.org>
Date: Thu, 26 May 2011 17:19:30 -0700
Subject: [PATCH] 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>
---
 .../dht/spi/util/AbstractWriteBuffer.java     | 25 +++++++++++++------
 1 file changed, 18 insertions(+), 7 deletions(-)

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 d40cbe31a..ad55206fe 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;
 	}