zimmermq opened a new issue, #909:
URL: https://github.com/apache/mina-sshd/issues/909
### Version
3.0.0-M4
### Bug description
`KexOutputHandler` uses a `ReentrantReadWriteLock`: `writeOrEnqueue()` takes
the **read** lock to decide-and-write a packet, while state changes and
`shutdown()` take the **write** lock via `updateState()`.
A single thread can deadlock itself:
1. An outgoing high-level packet (e.g. `SSH_MSG_CHANNEL_CLOSE` produced
while handling an inbound message) enters `KexOutputHandler.send()` →
`writeOrEnqueue()`, which acquires the **read lock** and, KEX being done,
performs the actual write via `filter.write()`.
2. The nio2 transport can invoke the write-completion handler **inline on
the calling thread** (`sun.nio.ch.Invoker.invokeDirect`). When that write
**fails** (peer already gone), the failure handler closes the session:
`Nio2Session.handleWriteCycleFailure` → `exceptionCaught` →
`AbstractSession.preClose` → `SshTransportFilter.shutdown` →
`KexFilter.shutdown()` → `KexOutputHandler.shutdown()`.
3. `KexOutputHandler.shutdown()` calls `updateState()`, which tries to
acquire the **write lock** — while the same thread still holds the **read
lock** from step 1. `ReentrantReadWriteLock` does not support upgrading, so the
thread parks forever holding its own read lock.
Because each `KexOutputHandler` has its own per-session lock,
`ThreadMXBean`/`jstack` reports **no deadlock**, but the nio2 worker is lost.
On a server with a small nio2 pool this consumes one worker per occurrence;
over time the pool is exhausted and the server accepts TCP connections but
never completes any new SSH key exchange (clients time out during handshake).
Other listeners (e.g. a separate FTP acceptor, an HTTP health endpoint) are
unaffected, which can mask the outage from health checks.
### How to reproduce
Deterministic unit test (JUnit 5 + Mockito) in package
`org.apache.sshd.common.session.filters.kex`. It models the inline
write-failure by having `filter.write()` invoke `output.shutdown()` on the
calling thread — exactly what `KexFilter.shutdown()` does at the end of the
real close chain (`output.shutdown()`), while `writeOrEnqueue()` still holds
the read lock:
```java
@Tag("NoIoTestCase")
class KexOutputHandlerDeadlockTest {
@Test
@Timeout(value = 30, unit = TimeUnit.SECONDS)
void readToWriteLockUpgradeSelfDeadlockOnInlineWriteFailure() throws
Exception {
KexFilter filter = mock(KexFilter.class);
AbstractSession session = mock(AbstractSession.class);
when(filter.getSession()).thenReturn(session);
// KEX finished: writeOrEnqueue() takes the direct-write branch
through filter.write().
when(filter.getKexState()).thenReturn(new
AtomicReference<>(KexState.DONE));
KexOutputHandler output = new KexOutputHandler(filter,
LoggerFactory.getLogger(KexOutputHandlerDeadlockTest.class));
// Model the nio2 write-failure callback firing inline on the
calling thread. The real chain is
// filter.write() -> ... -> Nio2Session.onFailed -> session close ->
SshTransportFilter.shutdown ->
// KexFilter.shutdown() -> KexOutputHandler.shutdown(). Invoke the
terminal call directly, on the
// same thread, while writeOrEnqueue() still holds the read lock.
when(filter.write(anyInt(),
any(Buffer.class))).thenAnswer(invocation -> {
output.shutdown();
return null;
});
Thread worker = new Thread(() -> {
try {
Buffer buffer = new ByteArrayBuffer(new byte[] { (byte)
SshConstants.SSH_MSG_CHANNEL_CLOSE });
output.send(SshConstants.SSH_MSG_CHANNEL_CLOSE, buffer);
} catch (IOException e) {
// not reached; the thread deadlocks before returning
}
}, "kex-deadlock-worker");
worker.setDaemon(true);
worker.start();
worker.join(TimeUnit.SECONDS.toMillis(5));
if (worker.isAlive()) {
String stack = Stream.of(worker.getStackTrace()).map(e -> "\tat
" + e).reduce("", (a, b) -> a + '\n' + b);
fail("KexOutputHandler.send() deadlocked (read-to-write lock
upgrade):" + stack);
}
}
}
```
On current `dev_3.0` this test **hangs and fails** (worker parked in
`ReentrantReadWriteLock$WriteLock.lock` via `shutdown()` → `updateState()`
while holding the read lock taken in `writeOrEnqueue()`).
### Actual behavior
The nio2 worker thread self-deadlocks: it holds the `KexOutputHandler` read
lock (from `writeOrEnqueue()`) and blocks forever trying to acquire the write
lock (from `shutdown()` → `updateState()`) on the same lock — an unsupported
read-to-write upgrade. The session never finishes closing and the worker is
never returned to the pool. With enough occurrences the server stops completing
new key exchanges while still accepting TCP connections.
### Expected behavior
A synchronous (inline) write failure during session teardown must not
deadlock. `KexOutputHandler.shutdown()` should complete regardless of whether
the calling thread already holds the read lock, and the nio2 worker should be
released.
### Relevant log output
```Shell
"sshd-SshServer[xxxxxxxx](port=9001)-nio2-thread-1" daemon prio=5 waiting on
condition
java.lang.Thread.State: WAITING (parking)
at jdk.internal.misc.Unsafe.park([email protected]/Native Method)
- parking to wait for <0x...> (a
java.util.concurrent.locks.ReentrantReadWriteLock$NonfairSync)
at
java.util.concurrent.locks.ReentrantReadWriteLock$WriteLock.lock([email protected]/ReentrantReadWriteLock.java:966)
at
org.apache.sshd.common.session.filters.kex.KexOutputHandler.updateState(KexOutputHandler.java:146)
at
org.apache.sshd.common.session.filters.kex.KexOutputHandler.shutdown(KexOutputHandler.java:193)
at
org.apache.sshd.common.session.filters.kex.KexFilter.shutdown(KexFilter.java:342)
at
org.apache.sshd.common.session.filters.SshTransportFilter.shutdown(SshTransportFilter.java:116)
at
org.apache.sshd.common.session.helpers.AbstractSession.preClose(AbstractSession.java:524)
at
org.apache.sshd.common.util.closeable.AbstractCloseable.close(AbstractCloseable.java:94)
at
org.apache.sshd.common.session.helpers.SessionHelper.exceptionCaught(SessionHelper.java:1029)
at
org.apache.sshd.common.session.helpers.AbstractSessionIoHandler.exceptionCaught(AbstractSessionIoHandler.java:65)
at
org.apache.sshd.common.io.nio2.Nio2Session.exceptionCaught(Nio2Session.java:208)
at
org.apache.sshd.common.io.nio2.Nio2Session.handleWriteCycleFailure(Nio2Session.java:598)
at
org.apache.sshd.common.io.nio2.Nio2Session$2.onFailed(Nio2Session.java:549)
at
org.apache.sshd.common.io.nio2.Nio2CompletionHandler.failed(Nio2CompletionHandler.java:42)
at sun.nio.ch.Invoker.invokeUnchecked([email protected]/Invoker.java:123)
at sun.nio.ch.Invoker.invokeDirect([email protected]/Invoker.java:140)
at
sun.nio.ch.UnixAsynchronousSocketChannelImpl.implWrite([email protected]/UnixAsynchronousSocketChannelImpl.java:753)
at
sun.nio.ch.AsynchronousSocketChannelImpl.write([email protected]/AsynchronousSocketChannelImpl.java:399)
at
org.apache.sshd.common.io.nio2.Nio2Session.doWriteCycle(Nio2Session.java:535)
at
org.apache.sshd.common.io.nio2.Nio2Session.startWriting(Nio2Session.java:516)
at
org.apache.sshd.common.io.nio2.Nio2Session.writeBuffer(Nio2Session.java:191)
at
org.apache.sshd.common.session.filters.CryptFilter$EncryptionHandler.send(CryptFilter.java:387)
at
org.apache.sshd.common.session.filters.CompressionFilter$Compressor.send(CompressionFilter.java:161)
at
org.apache.sshd.common.session.filters.kex.KexFilter$Sender.send(KexFilter.java:1300)
at
org.apache.sshd.common.session.filters.kex.KexFilter.write(KexFilter.java:1100)
at
org.apache.sshd.common.session.filters.kex.KexOutputHandler.writeOrEnqueue(KexOutputHandler.java:268)
<-- read lock held here
at
org.apache.sshd.common.session.filters.kex.KexOutputHandler.send(KexOutputHandler.java:225)
at
org.apache.sshd.common.session.helpers.AbstractSession.writePacket(AbstractSession.java:565)
at
org.apache.sshd.common.channel.AbstractChannel$GracefulChannelCloseable.close(AbstractChannel.java:646)
at
org.apache.sshd.common.session.helpers.AbstractConnectionService.channelClose(AbstractConnectionService.java:896)
at
org.apache.sshd.common.session.helpers.AbstractConnectionService.process(AbstractConnectionService.java:735)
at
org.apache.sshd.common.session.helpers.AbstractSession.handleMessage(AbstractSession.java:430)
... (inbound message processing; SSH_MSG_CHANNEL_CLOSE being written
back) ...
(Line numbers are from the 3.0.0-M4 runtime; the same method chain exists on
dev_3.0. `<0x...>` and the session id are redacted.)
```
### Other information
**Root cause.** `writeOrEnqueue()` deliberately holds the read lock across
`filter.write()` (see the comment there: it must decide-and-write atomically
w.r.t. KEX state so a concurrent `KEX_INIT` cannot reorder a high-level message
after our `KEX_INIT`). That is fine as long as the write cannot re-enter the
same lock — but the nio2 layer can run the write-failure callback inline, and
that callback's close path re-enters `KexOutputHandler.shutdown()` →
`updateState()`, which needs the write lock. Hence the illegal upgrade.
**Suggested fix (surgical).** In `shutdown()`, skip the write-lock
acquisition when the current thread already holds the read lock — the state
update is safe there because a held read lock already excludes all writers, and
the values involved are atomics:
```java
public void shutdown() {
shutDown.set(true);
Supplier<SimpleImmutableEntry<Integer, DefaultKeyExchangeFuture>> update
= () -> {
kexFlushed.set(true);
return new SimpleImmutableEntry<>(pendingPackets.size(),
kexFlushedFuture.get());
};
// A synchronous write failure in writeOrEnqueue() can close the session
- and thus call this
// shutdown() - inline on a thread that still holds the read lock.
Acquiring the write lock then would
// be an illegal read-to-write upgrade and would self-deadlock. Holding
the read lock already excludes
// writers, so in that reentrant case update the state without acquiring
the exclusive lock.
SimpleImmutableEntry<Integer, DefaultKeyExchangeFuture> items
= (lock.getReadHoldCount() > 0) ? update.get() :
updateState(update);
items.getValue().setValue(Boolean.valueOf(items.getKey().intValue() ==
0));
}
```
With this change the reproduction test above passes (the call returns in
well under a second). `shutdown()` is reached only via the one-time session
close (`AbstractCloseable.close()` uses `state.compareAndSet(Opened, ...)`),
and `DefaultSshFuture.setValue()` is already "set once" and thread-safe, so the
lock-free branch introduces no new race.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
