Merge "Make the FileLfsRepository thread safe"
diff --git a/org.eclipse.jgit.lfs.server.test/tst/org/eclipse/jgit/lfs/server/fs/UploadTest.java b/org.eclipse.jgit.lfs.server.test/tst/org/eclipse/jgit/lfs/server/fs/UploadTest.java
index 35bf09b..1fb91bd 100644
--- a/org.eclipse.jgit.lfs.server.test/tst/org/eclipse/jgit/lfs/server/fs/UploadTest.java
+++ b/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 @@
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);
+ }
+ }
}
diff --git a/org.eclipse.jgit.lfs.server/src/org/eclipse/jgit/lfs/server/fs/FileLfsRepository.java b/org.eclipse.jgit.lfs.server/src/org/eclipse/jgit/lfs/server/fs/FileLfsRepository.java
index 12da271..2e71c04 100644
--- a/org.eclipse.jgit.lfs.server/src/org/eclipse/jgit/lfs/server/fs/FileLfsRepository.java
+++ b/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 @@
private final String url;
private final Path dir;
- private AtomicObjectOutputStream out;
/**
* @param url
@@ -147,21 +144,11 @@
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) {
diff --git a/org.eclipse.jgit.lfs.server/src/org/eclipse/jgit/lfs/server/fs/ObjectUploadListener.java b/org.eclipse.jgit.lfs.server/src/org/eclipse/jgit/lfs/server/fs/ObjectUploadListener.java
index 05970ad..e524ac6 100644
--- a/org.eclipse.jgit.lfs.server/src/org/eclipse/jgit/lfs/server/fs/ObjectUploadListener.java
+++ b/org.eclipse.jgit.lfs.server/src/org/eclipse/jgit/lfs/server/fs/ObjectUploadListener.java
@@ -74,13 +74,13 @@
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 @@
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 @@
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 @@
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 @@
@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;
