Make the FileLfsRepository thread safe

The FileLfsRepository.out member could have been accessed from multiple
threads which would corrupt the content.

Don't store the AtomicObjectOutputStream in the FileLfsRepository.out but
move it to the ObjectUploadListener which is instantiated per-request.

Add a parallel upload test.

Change-Id: I62298630e99c46b500d376843ffcde934436215b
Signed-off-by: Saša Živkov <sasa.zivkov@sap.com>
Signed-off-by: Matthias Sohn <matthias.sohn@sap.com>

org.eclipse.jgit.lfs.server.test/tst/org/eclipse/jgit/lfs/server/fs/UploadTest.java

@@ -51,6 +51,13 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import org.eclipse.jgit.lfs.lib.AnyLongObjectId;
 import org.eclipse.jgit.lfs.lib.LongObjectId;
@@ -98,4 +105,36 @@ public void testLargeFileUpload() throws Exception {
 		assertEquals("expected object length " + Files.size(f), Files.size(f),
 				repository.getSize(id));
 	}
+
+	@Test
+	public void testParallelUploads() throws Exception {
+		int count = 10;
+		List<Path> paths = new ArrayList<>(count);
+
+		for (int i = 0; i < count; i++) {
+			Path f = Paths.get(getTempDirectory().toString(),
+					"largeRandomFile_" + i);
+			createPseudoRandomContentFile(f, 1 * MiB);
+			paths.add(f);
+		}
+
+		final CyclicBarrier barrier = new CyclicBarrier(count);
+
+		ExecutorService e = Executors.newFixedThreadPool(count);
+		try {
+			for (final Path p : paths) {
+				e.submit(new Callable<Void>() {
+					@Override
+					public Void call() throws Exception {
+						barrier.await();
+						putContent(p);
+						return null;
+					}
+				});
+			}
+		} finally {
+			e.shutdown();
+			e.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
+		}
+	}
 }

org.eclipse.jgit.lfs.server/src/org/eclipse/jgit/lfs/server/fs/FileLfsRepository.java

@@ -45,10 +45,8 @@
 import static org.eclipse.jgit.util.HttpSupport.HDR_AUTHORIZATION;
 
 import java.io.IOException;
-import java.nio.channels.Channels;
 import java.nio.channels.FileChannel;
 import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.WritableByteChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
@@ -70,7 +68,6 @@ public class FileLfsRepository implements LargeFileRepository {
 
 	private final String url;
 	private final Path dir;
-	private AtomicObjectOutputStream out;
 
 	/**
 	 * @param url
@@ -147,21 +144,11 @@ ReadableByteChannel getReadChannel(AnyLongObjectId id)
 		return FileChannel.open(getPath(id), StandardOpenOption.READ);
 	}
 
-	WritableByteChannel getWriteChannel(AnyLongObjectId id)
+	AtomicObjectOutputStream getOutputStream(AnyLongObjectId id)
 			throws IOException {
 		Path path = getPath(id);
 		Files.createDirectories(path.getParent());
-		out = new AtomicObjectOutputStream(path, id);
-		return Channels.newChannel(out);
-	}
-
-	/**
-	 * Abort the output stream
-	 */
-	void abortWrite() {
-		if (out != null) {
-			out.abort();
-		}
+		return new AtomicObjectOutputStream(path, id);
 	}
 
 	private static char[] toHexCharArray(int b) {

org.eclipse.jgit.lfs.server/src/org/eclipse/jgit/lfs/server/fs/ObjectUploadListener.java

@@ -74,13 +74,13 @@ class ObjectUploadListener implements ReadListener {
 
 	private final HttpServletResponse response;
 
-	private FileLfsRepository repository;
-
 	private final ServletInputStream in;
 
 	private final ReadableByteChannel inChannel;
 
-	private WritableByteChannel out;
+	private final AtomicObjectOutputStream out;
+
+	private WritableByteChannel channel;
 
 	private final ByteBuffer buffer = ByteBuffer.allocateDirect(8192);
 
@@ -98,12 +98,12 @@ public ObjectUploadListener(FileLfsRepository repository,
 			AsyncContext context, HttpServletRequest request,
 			HttpServletResponse response, AnyLongObjectId id)
 					throws FileNotFoundException, IOException {
-		this.repository = repository;
 		this.context = context;
 		this.response = response;
 		this.in = request.getInputStream();
 		this.inChannel = Channels.newChannel(in);
-		this.out = repository.getWriteChannel(id);
+		this.out = repository.getOutputStream(id);
+		this.channel = Channels.newChannel(out);
 		response.setContentType(Constants.CONTENT_TYPE_GIT_LFS_JSON);
 	}
 
@@ -117,12 +117,12 @@ public void onDataAvailable() throws IOException {
 		while (in.isReady()) {
 			if (inChannel.read(buffer) > 0) {
 				buffer.flip();
-				out.write(buffer);
+				channel.write(buffer);
 				buffer.compact();
 			} else {
 				buffer.flip();
 				while (buffer.hasRemaining()) {
-					out.write(buffer);
+					channel.write(buffer);
 				}
 				close();
 				return;
@@ -141,7 +141,7 @@ public void onAllDataRead() throws IOException {
 	protected void close() throws IOException {
 		try {
 			inChannel.close();
-			out.close();
+			channel.close();
 			// TODO check if status 200 is ok for PUT request, HTTP foresees 204
 			// for successful PUT without response body
 			response.setStatus(HttpServletResponse.SC_OK);
@@ -157,9 +157,9 @@ protected void close() throws IOException {
 	@Override
 	public void onError(Throwable e) {
 		try {
-			repository.abortWrite();
+			out.abort();
 			inChannel.close();
-			out.close();
+			channel.close();
 			int status;
 			if (e instanceof CorruptLongObjectException) {
 				status = HttpStatus.SC_BAD_REQUEST;