tomaswolf commented on code in PR #565:
URL: https://github.com/apache/mina-sshd/pull/565#discussion_r1693941698
##########
sshd-core/src/main/java/org/apache/sshd/core/CoreModuleProperties.java:
##########
@@ -800,6 +801,13 @@ public final class CoreModuleProperties {
}
});
+ /**
+ * Configuration value to enable {@code no-flow-control}. When set to
{@code true}, the option will be used if the
+ * connected peer also supports it. A value of {@code false} disables the
{@code no-flow-control} completely, while
+ * the default {@code null} value, will support it, but not request it.
+ */
+ public static final Property<Boolean> NO_FLOW_CONTROL =
Property.bool(NoFlowControl.NAME);
Review Comment:
_Where_ should this be set? For a server it's OK, set it on the `SshServer`.
But for the client side?
- On the `SshClient`? That's not useful if one uses a single `SshClient`for
multiple sessions; it would be advertised and might be activated for _all_
sessions created through that `SshClient`.
- On the `ClientSession`? That's way too late; a client session starts
immediately when instantiated, so if the user code want to set it on the
session, there'd be a race condition. The initial KEX might be on-going or
already over, and the client might already have passed the point where it would
send its SSH_MSG_EXT_INFO message.
##########
sshd-core/src/main/java/org/apache/sshd/common/channel/LocalWindow.java:
##########
@@ -70,6 +70,11 @@ public void consume(long len) throws IOException {
BufferUtils.validateUint32Value(len, "Invalid consumption length: %d");
checkInitialized("consume");
+ if (noFlowControl) {
+ // flow control is disabled, so just bail out
+ return;
+ }
+
Review Comment:
Additionally, `release()` must not do anything.
##########
sshd-core/src/test/java/org/apache/sshd/NoFlowControlTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sshd;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collection;
+import java.util.Deque;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.sshd.client.SshClient;
+import org.apache.sshd.client.channel.ClientChannel;
+import org.apache.sshd.client.channel.ClientChannelEvent;
+import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.io.IoInputStream;
+import org.apache.sshd.common.io.IoOutputStream;
+import org.apache.sshd.common.io.WritePendingException;
+import org.apache.sshd.common.util.buffer.Buffer;
+import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
+import org.apache.sshd.common.util.io.IoUtils;
+import org.apache.sshd.common.util.io.output.NoCloseOutputStream;
+import org.apache.sshd.common.util.logging.AbstractLoggingBean;
+import org.apache.sshd.common.util.threads.ThreadUtils;
+import org.apache.sshd.core.CoreModuleProperties;
+import org.apache.sshd.server.Environment;
+import org.apache.sshd.server.ExitCallback;
+import org.apache.sshd.server.SshServer;
+import org.apache.sshd.server.channel.ChannelSession;
+import org.apache.sshd.server.command.AsyncCommand;
+import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
+import org.apache.sshd.util.test.BaseTestSupport;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.FixMethodOrder;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+import org.junit.runners.MethodSorters;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test the no-flow-control extension.
+ */
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class NoFlowControlTest extends BaseTestSupport {
Review Comment:
This test simulates a download. There should also be a test showing a file
upload.
Plus: this needs tests with slow consumers, and it needs tests involving
port forwarding with slow consumers. I'm a bit worried that `no-flow-control`
may lead to a lot of data ending up being buffered somewhere. Especially if
it's done like in this test: I suspect the "pending queue" will then get very
large. That would be a no-go for a server, and might be a problem for a client.
If someone tunes a high latency connection by increasing send and receive
buffer sizes, might we get into trouble? (Especially on a server. Consider a
server with 1000 connections all having `no-flow-control` enabled and streaming
lots of data.)
##########
sshd-core/src/main/java/org/apache/sshd/common/session/helpers/KeyExchangeMessageHandler.java:
##########
@@ -236,7 +237,7 @@ public IoWriteFuture writePacket(Buffer buffer, long
timeout, TimeUnit unit) thr
int cmd = bufData[buffer.rpos()] & 0xFF;
boolean enqueued = false;
boolean isLowLevelMessage = cmd <= SshConstants.SSH_MSG_KEX_LAST &&
cmd != SshConstants.SSH_MSG_SERVICE_REQUEST
- && cmd != SshConstants.SSH_MSG_SERVICE_ACCEPT;
+ && cmd != SshConstants.SSH_MSG_SERVICE_ACCEPT && cmd !=
KexExtensions.SSH_MSG_EXT_INFO;
Review Comment:
I disagree with this. Yes, it's message number 7, so technically a
"transport layer generic messages" as RFC 4253 puts it. However, RFC 4253 was
written before RFC 8308, and RFC 4253 excludes already two messages from the
range 1-19.
Let's see: according to RFC 8308, a client may send SSH_MSG_EXT_INFO only
once immediately after its initial SSH_MSG_NEWKEYS. At that point, there is
definitely no KEX ongoing (even if the server required an immediate re-KEX by
sending another SSH_MSG_KEXINIT, that new KEX can only start once the client
has sent _its_ SSH_MSG_KEXINIT, which by definition must be after that
SSH_MSG_EXT_INFO message -- otherwise the client would send SSH_MSG_NEWKEYS -
SSH_MSG_KEXINIT - SSH_MSG_EXT_INFO, which is illegal according to RFC 8308.
A _server_ can likewise send SSH_MSG_EXT_INFO immediately after its initial
SSH_MSG_NEWKEYS. The same reasoning as above applies.
A server can also send the message immediately preceeding its
SSH_MSG_USERAUTH_SUCCESS. But that message cannot be sent while a KEX is
on-going; it would be delayed. So to send it immediately preceeding, the
SSH_MSG_EXT_INFO _must_ equally be delayed, otherwise one could get the
sequence <on-going KEX here> - SSH_MSG_EXT_INFO - SSH_MSG_NEWKEYS -
SSH_MSG_USERAUTH_SUCCESS, which also violates RFC 8308.
Therefore, SSH_MSG_EXT_INFO **must not** be treated as a "transport layer
generic message", and it must never be sent while a KEX is on-going.
The other message from RFC 8308, SSH_MSG_NEWCOMPRESS, also cannot be sent
during a KEX per section 3.2.1.
##########
sshd-core/src/main/java/org/apache/sshd/common/channel/RemoteWindow.java:
##########
@@ -67,6 +67,11 @@ public void consume(long len) {
BufferUtils.validateUint32Value(len, "Invalid consumption length: %d");
checkInitialized("consume");
+ if (noFlowControl) {
+ // flow control is disabled, so just bail out
+ return;
+ }
+
Review Comment:
Additionally, `waitAndConsume()` and `waitForSpace()`should also do an early
return. The should not wait for anything. `expand` must not do anything.
##########
sshd-core/src/main/java/org/apache/sshd/common/kex/extension/DefaultServerKexExtensionHandler.java:
##########
@@ -130,6 +136,23 @@ public void sendKexExtensions(Session session, KexPhase
phase) throws Exception
}
}
+ @Override
+ public boolean handleKexExtensionRequest(
+ Session session, int index, int count, String name, byte[] data)
+ throws IOException {
+ if (NoFlowControl.NAME.equals(name)) {
+ String o = NoFlowControl.INSTANCE.parseExtension(data);
+ Optional<Boolean> nfc =
CoreModuleProperties.NO_FLOW_CONTROL.get(session);
+ if (NoFlowControl.PREFERRED.equals(o) && nfc.orElse(Boolean.TRUE)
+ || NoFlowControl.SUPPORTED.equals(o) &&
nfc.orElse(Boolean.FALSE)) {
+ AbstractSession abstractSession
+ = ValidateUtils.checkInstanceOf(session,
AbstractSession.class, "Not a supported session: %s", session);
+ abstractSession.activateNoFlowControl();
+ }
+ }
Review Comment:
Can we avoid this code duplication between client & server?
##########
sshd-core/src/main/java/org/apache/sshd/common/channel/Window.java:
##########
@@ -94,6 +97,8 @@ protected void init(long size, long packetSize,
PropertyResolver resolver) {
}
synchronized (lock) {
+ Session session = channelInstance.getSession(); // this should
only be null during tests
+ this.noFlowControl = session != null && session.isNoFlowControl();
Review Comment:
Below:
```
this.maxSize = this.noFlowControl ? Integer.MAX_VALUE : size;
this.packetSize = packetSize;
updateSize(this.maxSize);
```
##########
sshd-core/src/main/java/org/apache/sshd/core/CoreModuleProperties.java:
##########
@@ -800,6 +801,13 @@ public final class CoreModuleProperties {
}
});
+ /**
+ * Configuration value to enable {@code no-flow-control}. When set to
{@code true}, the option will be used if the
+ * connected peer also supports it. A value of {@code false} disables the
{@code no-flow-control} completely, while
+ * the default {@code null} value, will support it, but not request it.
+ */
Review Comment:
I don't like the mis-use of Boolean for a tri-state value, and I don't like
the fact that `null` corresponds to `'s'` (supported).
Introduce an enum for this with two values: SUPPORTED and PREFERRED, and let
`null` mean "do not advertise `no-flow-control` at all".
Additionally: under what condition exactly would it make sense for a
_server_ to ever send `'p'` (PREFERRED)? A general-purpose server should
probably never do so and at most send `'s'`(SUPPORTED).
For a server the default should be `null` (don't advertise it). Users who
want to have a server that does implement this can set the property to
SUPPORTED explicitly.
For a client sending 's' probably makes not much sense (unless there is a
use case where having the server send 'p' would make sense). For a client it's
either 'p' ("yes, I want this if you can do it, too"), or don't advertise it at
all ("no, I don't want this, even if you can, because I might want to open
multiple simultaneous channels").
##########
sshd-core/src/test/java/org/apache/sshd/NoFlowControlTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sshd;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collection;
+import java.util.Deque;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.sshd.client.SshClient;
+import org.apache.sshd.client.channel.ClientChannel;
+import org.apache.sshd.client.channel.ClientChannelEvent;
+import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.io.IoInputStream;
+import org.apache.sshd.common.io.IoOutputStream;
+import org.apache.sshd.common.io.WritePendingException;
+import org.apache.sshd.common.util.buffer.Buffer;
+import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
+import org.apache.sshd.common.util.io.IoUtils;
+import org.apache.sshd.common.util.io.output.NoCloseOutputStream;
+import org.apache.sshd.common.util.logging.AbstractLoggingBean;
+import org.apache.sshd.common.util.threads.ThreadUtils;
+import org.apache.sshd.core.CoreModuleProperties;
+import org.apache.sshd.server.Environment;
+import org.apache.sshd.server.ExitCallback;
+import org.apache.sshd.server.SshServer;
+import org.apache.sshd.server.channel.ChannelSession;
+import org.apache.sshd.server.command.AsyncCommand;
+import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
+import org.apache.sshd.util.test.BaseTestSupport;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.FixMethodOrder;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+import org.junit.runners.MethodSorters;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test the no-flow-control extension.
+ */
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class NoFlowControlTest extends BaseTestSupport {
+
+ public static final byte END_FILE = '#';
+ public static final int BIG_MSG_SEND_COUNT = 10000;
+
+ @Rule
+ public Timeout timeout = Timeout.seconds(TimeUnit.MINUTES.toSeconds(6));
+
+ private SshServer sshServer;
+ private int port;
+
+ public NoFlowControlTest() {
+ super();
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ sshServer = setupTestServer();
+
+ byte[] msg =
IoUtils.toByteArray(getClass().getResourceAsStream("/big-msg.txt"));
Review Comment:
Where is that resource? Why use a resource at all?
##########
sshd-core/src/test/java/org/apache/sshd/NoFlowControlTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sshd;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collection;
+import java.util.Deque;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.sshd.client.SshClient;
+import org.apache.sshd.client.channel.ClientChannel;
+import org.apache.sshd.client.channel.ClientChannelEvent;
+import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.io.IoInputStream;
+import org.apache.sshd.common.io.IoOutputStream;
+import org.apache.sshd.common.io.WritePendingException;
+import org.apache.sshd.common.util.buffer.Buffer;
+import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
+import org.apache.sshd.common.util.io.IoUtils;
+import org.apache.sshd.common.util.io.output.NoCloseOutputStream;
+import org.apache.sshd.common.util.logging.AbstractLoggingBean;
+import org.apache.sshd.common.util.threads.ThreadUtils;
+import org.apache.sshd.core.CoreModuleProperties;
+import org.apache.sshd.server.Environment;
+import org.apache.sshd.server.ExitCallback;
+import org.apache.sshd.server.SshServer;
+import org.apache.sshd.server.channel.ChannelSession;
+import org.apache.sshd.server.command.AsyncCommand;
+import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
+import org.apache.sshd.util.test.BaseTestSupport;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.FixMethodOrder;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+import org.junit.runners.MethodSorters;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test the no-flow-control extension.
+ */
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class NoFlowControlTest extends BaseTestSupport {
+
+ public static final byte END_FILE = '#';
+ public static final int BIG_MSG_SEND_COUNT = 10000;
+
+ @Rule
+ public Timeout timeout = Timeout.seconds(TimeUnit.MINUTES.toSeconds(6));
+
+ private SshServer sshServer;
+ private int port;
+
+ public NoFlowControlTest() {
+ super();
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ sshServer = setupTestServer();
+
+ byte[] msg =
IoUtils.toByteArray(getClass().getResourceAsStream("/big-msg.txt"));
+ sshServer.setShellFactory(
+ channel -> new FloodingAsyncCommand(msg, BIG_MSG_SEND_COUNT,
END_FILE));
+
+ sshServer.setKeyPairProvider(new SimpleGeneratorHostKeyProvider());
+ sshServer.start();
+ port = sshServer.getPort();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (sshServer != null) {
+ sshServer.stop();
+ sshServer.close(true);
+ }
+ }
+
+ @Test
+ public void testNoFlowControl() throws Exception {
+ try (SshClient client = setupTestClient()) {
+ CoreModuleProperties.NO_FLOW_CONTROL.set(client, true);
+ client.start();
+
+ try (ClientSession session = client.connect(getCurrentTestName(),
TEST_LOCALHOST, port)
+ .verify(CONNECT_TIMEOUT).getSession()) {
+ session.addPasswordIdentity(getCurrentTestName());
+ session.auth().verify(AUTH_TIMEOUT);
+
+ try (ClientChannel channel = session.createShellChannel()) {
+ channel.setOut(new VerifyingOutputStream(channel,
END_FILE));
+ channel.setErr(new NoCloseOutputStream(System.err));
+ channel.open().verify(OPEN_TIMEOUT);
+
+ assertTrue(session.isNoFlowControl());
+
assertTrue(sshServer.getActiveSessions().get(0).isNoFlowControl());
+
+ Collection<ClientChannelEvent> result
+ =
channel.waitFor(EnumSet.of(ClientChannelEvent.CLOSED),
TimeUnit.MINUTES.toMillis(2L));
+ assertFalse("Timeout while waiting for channel closure",
result.contains(ClientChannelEvent.TIMEOUT));
+ }
+ } finally {
+ client.stop();
+ }
+ }
+ }
+
+ /**
+ * Read all incoming data and if END_FILE symbol is detected, kill client
session to end test
Review Comment:
Doesn't kill the client session; it closes the client channel (hard, at
that). Why would the client do so? Wouldn't it be more natural to have the
server close the channel (gracefully!) after it has sent the last data?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscr...@mina.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscr...@mina.apache.org
For additional commands, e-mail: dev-h...@mina.apache.org
