Mirrors/sshj
1
0
Fork 0
mirror of https://github.com/hierynomus/sshj.git synced 2025-12-07 15:50:57 +03:00
Code Issues Actions Packages Projects Releases Wiki Activity

Implement read ahead to speed up transfer rates for downloads by a magnitude.

Browse Source
This commit is contained in:
David Kocher
2014-05-09 13:43:52 +02:00
parent 93f1543af8
commit f354fd6661
1 changed files with 73 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
src/main/java/net/schmizz/sshj/sftp/RemoteFile.java
View File

@@ -15,9 +15,6 @@
*/ */
package net.schmizz.sshj.sftp; package net.schmizz.sshj.sftp;
import net.schmizz.concurrent.Promise;
import net.schmizz.sshj.sftp.Response.StatusCode;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.io.OutputStream; import java.io.OutputStream;
@@ -25,6 +22,10 @@ import java.util.LinkedList;
import java.util.Queue; import java.util.Queue;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import net.schmizz.concurrent.Promise;
import net.schmizz.sshj.common.Buffer;
import net.schmizz.sshj.sftp.Response.StatusCode;
public class RemoteFile public class RemoteFile
extends RemoteResource { extends RemoteResource {
@@ -52,9 +53,16 @@ public class RemoteFile
public int read(long fileOffset, byte[] to, int offset, int len) public int read(long fileOffset, byte[] to, int offset, int len)
throws IOException { throws IOException {
final Response res = requester.request( final Response res = this.asyncRead(fileOffset, len).retrieve(requester.getTimeoutMs(), TimeUnit.MILLISECONDS);
newRequest(PacketType.READ).putUInt64(fileOffset).putUInt32(len) return this.checkReadResponse(res, to, offset);
).retrieve(requester.getTimeoutMs(), TimeUnit.MILLISECONDS); }
protected Promise<Response, SFTPException> asyncRead(long fileOffset, int len)
throws IOException {
return requester.request(newRequest(PacketType.READ).putUInt64(fileOffset).putUInt32(len));
}
protected int checkReadResponse(Response res, byte[] to, int offset) throws Buffer.BufferException, SFTPException {
switch(res.getType()) { switch(res.getType()) {
case DATA: case DATA:
int recvLen = res.readUInt32AsInt(); int recvLen = res.readUInt32AsInt();
@@ -72,7 +80,7 @@ public class RemoteFile
public void write(long fileOffset, byte[] data, int off, int len) public void write(long fileOffset, byte[] data, int off, int len)
throws IOException { throws IOException {
checkResponse(asyncWrite(fileOffset, data, off, len)); checkWriteResponse(asyncWrite(fileOffset, data, off, len));
} }
protected Promise<Response, SFTPException> asyncWrite(long fileOffset, byte[] data, int off, int len) protected Promise<Response, SFTPException> asyncWrite(long fileOffset, byte[] data, int off, int len)
@@ -84,7 +92,7 @@ public class RemoteFile
); );
} }
private void checkResponse(Promise<Response, SFTPException> responsePromise) private void checkWriteResponse(Promise<Response, SFTPException> responsePromise)
throws SFTPException { throws SFTPException {
responsePromise.retrieve(requester.getTimeoutMs(), TimeUnit.MILLISECONDS).ensureStatusPacketIsOK(); responsePromise.retrieve(requester.getTimeoutMs(), TimeUnit.MILLISECONDS).ensureStatusPacketIsOK();
} }
@@ -140,7 +148,7 @@ public class RemoteFile
public void write(byte[] buf, int off, int len) public void write(byte[] buf, int off, int len)
throws IOException { throws IOException {
if(unconfirmedWrites.size() > maxUnconfirmedWrites) { if(unconfirmedWrites.size() > maxUnconfirmedWrites) {
checkResponse(unconfirmedWrites.remove()); checkWriteResponse(unconfirmedWrites.remove());
} }
unconfirmedWrites.add(RemoteFile.this.asyncWrite(fileOffset, buf, off, len)); unconfirmedWrites.add(RemoteFile.this.asyncWrite(fileOffset, buf, off, len));
fileOffset += len; fileOffset += len;
@@ -150,7 +158,7 @@ public class RemoteFile
public void flush() public void flush()
throws IOException { throws IOException {
while(!unconfirmedWrites.isEmpty()) { while(!unconfirmedWrites.isEmpty()) {
checkResponse(unconfirmedWrites.remove()); checkWriteResponse(unconfirmedWrites.remove());
} }
} }
@@ -167,6 +175,11 @@ public class RemoteFile
private final byte[] b = new byte[1]; private final byte[] b = new byte[1];
private final int maxUnconfirmedReads;
private final Queue<Promise<Response, SFTPException>> unconfirmedReads;
private boolean eof;
private long fileOffset; private long fileOffset;
private long markPos; private long markPos;
private long readLimit; private long readLimit;
@@ -176,7 +189,13 @@ public class RemoteFile
} }
public RemoteFileInputStream(long fileOffset) { public RemoteFileInputStream(long fileOffset) {
this(fileOffset, 0);
}
public RemoteFileInputStream(long fileOffset, int maxUnconfirmedReads) {
this.fileOffset = fileOffset; this.fileOffset = fileOffset;
this.maxUnconfirmedReads = maxUnconfirmedReads;
this.unconfirmedReads = new LinkedList<Promise<Response, SFTPException>>();
} }
@Override @Override
@@ -187,7 +206,7 @@ public class RemoteFile
@Override @Override
public void mark(int readLimit) { public void mark(int readLimit) {
this.readLimit = readLimit; this.readLimit = readLimit;
markPos = fileOffset; this.markPos = fileOffset;
} }
@Override @Override
@@ -211,14 +230,45 @@ public class RemoteFile
@Override @Override
public int read(byte[] into, int off, int len) public int read(byte[] into, int off, int len)
throws IOException { throws IOException {
int read = RemoteFile.this.read(fileOffset, into, off, len); while(!eof && unconfirmedReads.size() <= maxUnconfirmedReads) {
if (read != -1) { // Send read requests as long as there is no EOF and we have not reached the maximum parallelism
fileOffset += read; unconfirmedReads.add(asyncRead(fileOffset, len));
if (markPos != 0 && read > readLimit) // Invalidate mark position fileOffset += len;
}
if(unconfirmedReads.isEmpty()) {
// Attempted to read while status was already received
return -1;
}
// Retrieve first in
final Response res = unconfirmedReads.remove().retrieve(
requester.getTimeoutMs(), TimeUnit.MILLISECONDS);
final int recvLen = checkReadResponse(res, into, off);
if(markPos != 0 && recvLen > readLimit) // Invalidate mark position
{
markPos = 0; markPos = 0;
} }
return read; if(-1 == recvLen) {
eof = true;
}
return recvLen;
} }
@Override
public void close() throws IOException {
while(!unconfirmedReads.isEmpty()) {
final Response res = unconfirmedReads.remove().retrieve(
requester.getTimeoutMs(), TimeUnit.MILLISECONDS);
switch(res.getType()) {
case STATUS:
res.ensureStatusIs(StatusCode.EOF);
break;
case DATA:
log.warn("Pending data packet from read response discarded");
continue;
default:
throw new SFTPException("Unexpected packet: " + res.getType());
}
}
}
} }
} }