You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mina.apache.org by tw...@apache.org on 2024/08/06 19:30:13 UTC
(mina-sshd) 01/03: GH-524: Avoid unnecessary work in SFTP uploads
This is an automated email from the ASF dual-hosted git repository.
twolf pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mina-sshd.git
commit c2a476094e2119971aa07be457e8a0aa1c0a3ec6
Author: Thomas Wolf <tw...@apache.org>
AuthorDate: Tue Aug 6 20:15:51 2024 +0200
GH-524: Avoid unnecessary work in SFTP uploads
Streamline ACK checking a bit. Do a quick check for the OK case first
and do the full and proper parsing only if the status is not OK. Also
avoid getting the IDLE_TIMEOUT repeatedly when draining remaining ACKs
when SftpOutputStreamAsync is closed.
---
.../sftp/client/impl/SftpOutputStreamAsync.java | 39 +++++++++++++++++++---
.../apache/sshd/sftp/client/impl/SftpResponse.java | 3 +-
.../apache/sshd/sftp/client/impl/SftpStatus.java | 8 ++++-
3 files changed, 44 insertions(+), 6 deletions(-)
diff --git a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpOutputStreamAsync.java b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpOutputStreamAsync.java
index bd9200083..c35c05d6e 100644
--- a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpOutputStreamAsync.java
+++ b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpOutputStreamAsync.java
@@ -32,16 +32,19 @@ import java.util.Objects;
import org.apache.sshd.common.SshConstants;
import org.apache.sshd.common.session.Session;
import org.apache.sshd.common.session.helpers.PacketBuffer;
+import org.apache.sshd.common.util.GenericUtils;
import org.apache.sshd.common.util.ValidateUtils;
import org.apache.sshd.common.util.buffer.Buffer;
import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
import org.apache.sshd.common.util.io.output.OutputStreamWithChannel;
+import org.apache.sshd.core.CoreModuleProperties;
import org.apache.sshd.sftp.client.SftpClient;
import org.apache.sshd.sftp.client.SftpClient.CloseableHandle;
import org.apache.sshd.sftp.client.SftpClient.OpenMode;
import org.apache.sshd.sftp.client.SftpClientHolder;
import org.apache.sshd.sftp.client.SftpMessage;
import org.apache.sshd.sftp.common.SftpConstants;
+import org.apache.sshd.sftp.common.SftpException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -327,8 +330,7 @@ public class SftpOutputStreamAsync extends OutputStreamWithChannel implements Sf
}
pendingAcks.removeFirst();
- SftpResponse response = SftpResponse.parse(SftpConstants.SSH_FXP_WRITE, buf);
- client.checkResponseStatus(SftpConstants.SSH_FXP_WRITE, ack.id, SftpStatus.parse(response));
+ checkStatus(client, buf);
}
if (buffer == null) {
@@ -366,6 +368,27 @@ public class SftpOutputStreamAsync extends OutputStreamWithChannel implements Sf
buffer = null;
}
+ private void checkStatus(AbstractSftpClient client, Buffer buf) throws IOException {
+ if (buf.available() >= 13) {
+ int rpos = buf.rpos();
+ buf.rpos(rpos + 4); // Skip length
+ int cmd = buf.getUByte();
+ if (cmd != SftpConstants.SSH_FXP_STATUS) {
+ throw new SftpException(SftpConstants.SSH_FX_BAD_MESSAGE,
+ "Unexpected SFTP response; expected SSH_FXP_STATUS but got "
+ + SftpConstants.getCommandMessageName(cmd));
+ }
+ buf.rpos(rpos + 9); // Skip ahead until after the id
+ if (buf.getInt() == SftpConstants.SSH_FX_OK) {
+ return;
+ }
+ // Reset and do the full parse
+ buf.rpos(rpos);
+ }
+ SftpResponse response = SftpResponse.parse(SftpConstants.SSH_FXP_WRITE, buf);
+ client.checkResponseStatus(SftpConstants.SSH_FXP_WRITE, response.getId(), SftpStatus.parse(response));
+ }
+
@Override
public void close() throws IOException {
if (!isOpen()) {
@@ -388,6 +411,10 @@ public class SftpOutputStreamAsync extends OutputStreamWithChannel implements Sf
lastMsg = null;
}
+ Duration idleTimeout = CoreModuleProperties.IDLE_TIMEOUT.getRequired(getClient().getClientSession());
+ if (GenericUtils.isNegativeOrNull(idleTimeout)) {
+ idleTimeout = CoreModuleProperties.IDLE_TIMEOUT.getRequiredDefault();
+ }
AbstractSftpClient client = getClient();
for (int ackIndex = 1; !pendingAcks.isEmpty(); ackIndex++) {
SftpAckData ack = pendingAcks.removeFirst();
@@ -395,11 +422,15 @@ public class SftpOutputStreamAsync extends OutputStreamWithChannel implements Sf
log.debug("close({}) processing ack #{}: {}", this, ackIndex, ack);
}
- SftpResponse response = client.response(SftpConstants.SSH_FXP_WRITE, ack.id);
+ Buffer buf = client.receive(ack.id, idleTimeout);
+ if (buf == null) {
+ log.debug("close({}) no ack response for {}", this, ack);
+ break;
+ }
if (debugEnabled) {
log.debug("close({}) processing ack #{} response for {}", this, ackIndex, ack);
}
- client.checkResponseStatus(SftpConstants.SSH_FXP_WRITE, response.getId(), SftpStatus.parse(response));
+ checkStatus(client, buf);
}
} finally {
if (ownsHandle) {
diff --git a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpResponse.java b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpResponse.java
index fd8787d9f..3ce0839fc 100644
--- a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpResponse.java
+++ b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpResponse.java
@@ -71,7 +71,8 @@ public final class SftpResponse {
int length = buffer.getInt();
int type = buffer.getUByte();
int id = buffer.getInt();
- validateIncomingResponse(cmd, id, type, length, buffer);
+ // No need to validate the length here: the way we assemble these buffers guarantees that
+ // the length is reasonable and does not exceed buffer.available().
return new SftpResponse(cmd, id, type, length, buffer);
}
diff --git a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpStatus.java b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpStatus.java
index 5d1511fcf..93fea737d 100644
--- a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpStatus.java
+++ b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/impl/SftpStatus.java
@@ -20,6 +20,7 @@ package org.apache.sshd.sftp.client.impl;
import org.apache.sshd.common.util.buffer.Buffer;
import org.apache.sshd.sftp.common.SftpConstants;
+import org.apache.sshd.sftp.common.SftpException;
/**
* A representation of a SSH_FXP_STATUS record.
@@ -69,7 +70,12 @@ public final class SftpStatus {
return new SftpStatus(code, message, language);
}
- public static SftpStatus parse(SftpResponse response) {
+ public static SftpStatus parse(SftpResponse response) throws SftpException {
+ if (response.getType() != SftpConstants.SSH_FXP_STATUS) {
+ throw new SftpException(
+ SftpConstants.SSH_FX_BAD_MESSAGE, "Unexpected SFTP response: expected SSH_FXP_STATUS but got "
+ + SftpConstants.getCommandMessageName(response.getType()));
+ }
return parse(response.getBuffer());
}
}