[jira] [Comment Edited] (SSHD-721) deadlock: all nio workers wait to be woken up
[
https://issues.apache.org/jira/browse/SSHD-721?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16438803#comment-16438803
]
Guillaume Nodet edited comment on SSHD-721 at 4/15/18 6:58 PM:
---
I think the code needs to be fixed: you can't perform a wait on the channel
from a worker thread or you risk hitting this very issue: i.e. thread pool
exhaustion.
Could you try the following patch :
{code}
diff --git
a/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
b/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
index 79364152..751f73a3 100644
---
a/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
+++
b/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
@@ -37,6 +37,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.sshd.client.channel.ClientChannelEvent;
+import org.apache.sshd.client.future.OpenFuture;
import org.apache.sshd.common.Closeable;
import org.apache.sshd.common.Factory;
import org.apache.sshd.common.FactoryManager;
@@ -979,15 +980,22 @@ public class DefaultForwardingFilter
session, channel, totalMessages,
message.available());
}
-Collection result =
channel.waitFor(STATIC_IO_MSG_RECEIVED_EVENTS, Long.MAX_VALUE);
-if (traceEnabled) {
-log.trace("messageReceived({}) channel={}, count={}, len={}
wait result: {}",
- session, channel, totalMessages,
message.available(), result);
+OpenFuture future = channel.getOpenFuture();
+if (future.isOpened()) {
+OutputStream outputStream = channel.getInvertedIn();
+outputStream.write(buffer.array(), buffer.rpos(),
buffer.available());
+outputStream.flush();
+} else {
+future.addListener(f -> {
+try {
+OutputStream outputStream = channel.getInvertedIn();
+outputStream.write(buffer.array(), buffer.rpos(),
buffer.available());
+outputStream.flush();
+} catch (IOException e) {
+channel.getSession().exceptionCaught(e);
+}
+});
}
-
-OutputStream outputStream = channel.getInvertedIn();
-outputStream.write(buffer.array(), buffer.rpos(),
buffer.available());
-outputStream.flush();
}
@Override
{code}
was (Author: gnt):
I think the code needs to be fixed: you can't perform an infinite wait on the
channel from a worker thread or you'll run into this very issue: i.e. thread
pool exhaustion.
Could you try the following patch :
{code}
diff --git
a/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
b/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
index 79364152..751f73a3 100644
---
a/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
+++
b/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
@@ -37,6 +37,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.sshd.client.channel.ClientChannelEvent;
+import org.apache.sshd.client.future.OpenFuture;
import org.apache.sshd.common.Closeable;
import org.apache.sshd.common.Factory;
import org.apache.sshd.common.FactoryManager;
@@ -979,15 +980,22 @@ public class DefaultForwardingFilter
session, channel, totalMessages,
message.available());
}
-Collection result =
channel.waitFor(STATIC_IO_MSG_RECEIVED_EVENTS, Long.MAX_VALUE);
-if (traceEnabled) {
-log.trace("messageReceived({}) channel={}, count={}, len={}
wait result: {}",
- session, channel, totalMessages,
message.available(), result);
+OpenFuture future = channel.getOpenFuture();
+if (future.isOpened()) {
+OutputStream outputStream = channel.getInvertedIn();
+outputStream.write(buffer.array(), buffer.rpos(),
buffer.available());
+outputStream.flush();
+} else {
+future.addListener(f -> {
+try {
+OutputStream outputStream = channel.getInvertedIn();
+outputStream.write(buffer.array(), buffer.rpos(),
buffer.available());
+outputStream.flush();
+} catch (IOException e) {
+channel.getSession().exceptionCaught(e);
+}
+});
}
-
-OutputStream
[jira] [Commented] (SSHD-721) deadlock: all nio workers wait to be woken up
[
https://issues.apache.org/jira/browse/SSHD-721?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16438803#comment-16438803
]
Guillaume Nodet commented on SSHD-721:
--
I think the code needs to be fixed: you can't perform an infinite wait on the
channel from a worker thread or you'll run into this very issue: i.e. thread
pool exhaustion.
Could you try the following patch :
{code}
diff --git
a/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
b/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
index 79364152..751f73a3 100644
---
a/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
+++
b/sshd-core/src/main/java/org/apache/sshd/common/forward/DefaultForwardingFilter.java
@@ -37,6 +37,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.sshd.client.channel.ClientChannelEvent;
+import org.apache.sshd.client.future.OpenFuture;
import org.apache.sshd.common.Closeable;
import org.apache.sshd.common.Factory;
import org.apache.sshd.common.FactoryManager;
@@ -979,15 +980,22 @@ public class DefaultForwardingFilter
session, channel, totalMessages,
message.available());
}
-Collection result =
channel.waitFor(STATIC_IO_MSG_RECEIVED_EVENTS, Long.MAX_VALUE);
-if (traceEnabled) {
-log.trace("messageReceived({}) channel={}, count={}, len={}
wait result: {}",
- session, channel, totalMessages,
message.available(), result);
+OpenFuture future = channel.getOpenFuture();
+if (future.isOpened()) {
+OutputStream outputStream = channel.getInvertedIn();
+outputStream.write(buffer.array(), buffer.rpos(),
buffer.available());
+outputStream.flush();
+} else {
+future.addListener(f -> {
+try {
+OutputStream outputStream = channel.getInvertedIn();
+outputStream.write(buffer.array(), buffer.rpos(),
buffer.available());
+outputStream.flush();
+} catch (IOException e) {
+channel.getSession().exceptionCaught(e);
+}
+});
}
-
-OutputStream outputStream = channel.getInvertedIn();
-outputStream.write(buffer.array(), buffer.rpos(),
buffer.available());
-outputStream.flush();
}
@Override
{code}
> deadlock: all nio workers wait to be woken up
> -
>
> Key: SSHD-721
> URL: https://issues.apache.org/jira/browse/SSHD-721
> Project: MINA SSHD
> Issue Type: Bug
>Affects Versions: 1.3.0, 1.4.0
>Reporter: Markus Rathgeb
>Priority: Major
>
> I am using sshd-core for a server machine (S) that accepts incoming
> connections and port forwarding requests.
> There are client machines (C) that run servers that should be accessible by a
> tunnel to the server.
> On the client machines (C) also an implementation using sshd-core is running
> that establish the connection to the server (S) and initiate the port
> forwarding.
> Other clients are using the tunnelled connection to communicate with the
> servers that are running on the client machines (C).
> Sometimes I realized that no data is transferred anymore (through the
> tunnels).
> All the worker reside in the waitFor function and no one wakes them up.
> {noformat}
> "sshd-SshServer[67de991c]-nio2-thread-3" - Thread t@125
>java.lang.Thread.State: TIMED_WAITING
> at java.lang.Object.wait(Native Method)
> - waiting on <132c6b60> (a java.lang.Object)
> at
> org.apache.sshd.client.channel.AbstractClientChannel.waitFor(AbstractClientChannel.java:244)
> at
> org.apache.sshd.common.forward.DefaultTcpipForwarder$StaticIoHandler.messageReceived(DefaultTcpipForwarder.java:984)
> at
> org.apache.sshd.common.io.nio2.Nio2Session.handleReadCycleCompletion(Nio2Session.java:276)
> at
> org.apache.sshd.common.io.nio2.Nio2Session$1.onCompleted(Nio2Session.java:256)
> at
> org.apache.sshd.common.io.nio2.Nio2Session$1.onCompleted(Nio2Session.java:253)
> at
> org.apache.sshd.common.io.nio2.Nio2CompletionHandler.lambda$completed$0(Nio2CompletionHandler.java:38)
> at
> org.apache.sshd.common.io.nio2.Nio2CompletionHandler$$Lambda$45/1071326492.run(Unknown
> Source)
> at java.security.AccessController.doPrivileged(Native Method)
> at
> org.apache.sshd.common.io.nio2.Nio2CompletionHandler.completed(Nio2CompletionHandler.java:37)
> at sun.nio.ch.Invoker.invokeUnchecked(Invoker.java:126)
> at sun.nio.ch.Invoker$2.run(Invoker.java:218)
>
