Browse Source

Fix missing flush in StreamCopyThread

It is possible that StreamCopyThread will not flush everything
from it's src to it's dst.  In most cases StreamCopyThread works
like this:

  in loop:
    n = src.read(buf);
    dst.write(buf, 0, n);

and when we want to flush, we interrupt() StreamCopyThread and it
flushes everything it wrote to dst.

The problem is that our interrupt() could interrupt reading. In this
case we will flush everything we wrote to dst, but not everything
we wrote to src.

Change-Id: Ifaf4d8be87535c7364dd59b217dfc631460018ff
Signed-off-by: Shawn O. Pearce <spearce@spearce.org>
stable-0.9
Dmitry Neverov 15 years ago committed by Shawn O. Pearce
parent
commit
44854741c5
  1. 26
      org.eclipse.jgit/src/org/eclipse/jgit/util/io/StreamCopyThread.java

26
org.eclipse.jgit/src/org/eclipse/jgit/util/io/StreamCopyThread.java

@ -47,7 +47,6 @@ import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.io.InterruptedIOException; import java.io.InterruptedIOException;
import java.io.OutputStream; import java.io.OutputStream;
import java.util.concurrent.atomic.AtomicInteger;
/** Thread to copy from an input stream to an output stream. */ /** Thread to copy from an input stream to an output stream. */
public class StreamCopyThread extends Thread { public class StreamCopyThread extends Thread {
@ -57,8 +56,6 @@ public class StreamCopyThread extends Thread {
private final OutputStream dst; private final OutputStream dst;
private final AtomicInteger flushCounter = new AtomicInteger(0);
private volatile boolean done; private volatile boolean done;
/** /**
@ -85,7 +82,6 @@ public class StreamCopyThread extends Thread {
* the request. * the request.
*/ */
public void flush() { public void flush() {
flushCounter.incrementAndGet();
interrupt(); interrupt();
} }
@ -113,10 +109,13 @@ public class StreamCopyThread extends Thread {
public void run() { public void run() {
try { try {
final byte[] buf = new byte[BUFFER_SIZE]; final byte[] buf = new byte[BUFFER_SIZE];
int interruptCounter = 0;
for (;;) { for (;;) {
try { try {
if (needFlush()) if (interruptCounter > 0) {
dst.flush(); dst.flush();
interruptCounter--;
}
if (done) if (done)
break; break;
@ -125,17 +124,25 @@ public class StreamCopyThread extends Thread {
try { try {
n = src.read(buf); n = src.read(buf);
} catch (InterruptedIOException wakey) { } catch (InterruptedIOException wakey) {
interruptCounter++;
continue; continue;
} }
if (n < 0) if (n < 0)
break; break;
boolean writeInterrupted = false;
for (;;) { for (;;) {
try { try {
dst.write(buf, 0, n); dst.write(buf, 0, n);
} catch (InterruptedIOException wakey) { } catch (InterruptedIOException wakey) {
writeInterrupted = true;
continue; continue;
} }
// set interrupt status, which will be checked
// when we block in src.read
if (writeInterrupted)
interrupt();
break; break;
} }
} catch (IOException e) { } catch (IOException e) {
@ -155,13 +162,4 @@ public class StreamCopyThread extends Thread {
} }
} }
} }
private boolean needFlush() {
int i = flushCounter.get();
if (i > 0) {
flushCounter.decrementAndGet();
return true;
}
return false;
}
} }

Loading…
Cancel
Save