This is an automated email from the ASF dual-hosted git repository.
chrisdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new 7492d5d5a2 feat(transport-tcp): replace NIO selector with
per-connection virtual-thread blocking I/O (#2612)
7492d5d5a2 is described below
commit 7492d5d5a2b3027463d9e742704592920cc30fc1
Author: LivingLikeKrillin <[email protected]>
AuthorDate: Thu Jun 25 17:22:33 2026 +0900
feat(transport-tcp): replace NIO selector with per-connection
virtual-thread blocking I/O (#2612)
* feat(transport-tcp): replace NIO selector with per-connection
virtual-thread blocking I/O
Each connection runs a blocking SocketChannel.read() loop on its own
virtual thread
instead of a per-connection NIO Selector. On Java 21 blocking-mode
reads/writes park the
virtual thread and release the carrier, so the selector (which pins the
carrier in
select()) and the OP_WRITE + Thread.sleep(1) write busy-wait are removed. A
full ring
buffer applies backpressure (park-and-retry) instead of toggling OP_READ.
Public surface, readLock, RingBuffer, and the AsyncTransportInstance
callback contract are
unchanged: the existing TcpTransportInstanceTest (31 tests) passes
unmodified.
Scaling probe (TcpTransportInstanceScalingTest): 200 idle connections use 2
carrier
threads with the blocking model vs 201 with the selector model.
* fix(transport-tcp): emit close log/audit only on successful close
Move the "TCP connection closed" debug line and CLOSE audit event out of
the finally block and into the success path of close(). Previously they
ran even when socketChannel.close() threw and the method rethrew, so a
failed close logged both an ERROR audit event and a misleading CLOSE
"Closed" event. The readThread.join() stays in finally so the read loop
is always awaited. Also correct an inaccurate comment in the scaling
test (the stop-flag holder is a field, not a way to avoid one).
* fix(transport-tcp): drop dead write -1 check and tidy audit/test nits
- write(): blocking SocketChannel.write() never returns -1 (that signals
read EOF), so the `written == -1` branch was dead code. A broken or closed
connection already surfaces as IOException/AsynchronousCloseException,
both
handled below. Remove the check.
- constructor: errorMsg already embeds e.getMessage(), so the second ERROR
audit event duplicated the first. Emit a single event.
- constructor: start the read-loop virtual thread last (after the INFO log
and CONNECT audit), so an unchecked throw from logging/audit cannot leak
an
already-running read thread and the open SocketChannel — the catch only
handles IOException and does not stop the read loop.
- close(): skip readThread.join() when close() runs on the read thread
itself
(a disconnect/data listener calling close()), since joining yourself only
stalls for the timeout and the loop already exits once open is false.
- scaling test: take one Thread.getAllStackTraces() snapshot so carriers and
total are counted from the same instant instead of two separate calls.
- scaling test: exclude ForkJoinPool.commonPool workers from the carrier
count so unrelated parallel-stream workers cannot inflate it.
---
.../java/transport/tcp/TcpTransportInstance.java | 329 ++++++++-------------
.../tcp/TcpTransportInstanceScalingTest.java | 127 ++++++++
2 files changed, 253 insertions(+), 203 deletions(-)
diff --git
a/plc4j/transports/tcp/src/main/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstance.java
b/plc4j/transports/tcp/src/main/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstance.java
index 96e2df38cc..6b67613c05 100644
---
a/plc4j/transports/tcp/src/main/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstance.java
+++
b/plc4j/transports/tcp/src/main/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstance.java
@@ -33,16 +33,22 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
+import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.SocketChannel;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
/**
- * Java 21+ optimized version using virtual threads - TCP transport
implementation using NIO SocketChannel with async support.
- * Implements AsyncTransportInstance for event-driven I/O without polling.
+ * Java 21+ TCP transport using one virtual thread per connection doing
blocking
+ * {@link SocketChannel} reads/writes. On Java 21 a virtual thread blocked in
a blocking-mode
+ * channel read/write parks and releases its carrier (the JDK registers the fd
with the NIO
+ * poller), so there is no NIO {@link java.nio.channels.Selector}, no
readiness-event loop, and
+ * no busy-wait. The public surface and the {@link AsyncTransportInstance}
callback contract are
+ * unchanged: the read loop fills the {@link RingBuffer} (under {@code
readLock}) and invokes the
+ * registered data listener exactly as the previous selector loop did.
*/
public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConfiguration> implements
AsyncTransportInstance<TcpTransportConfiguration> {
@@ -52,22 +58,21 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
private final SocketChannel socketChannel;
private final RingBuffer ringBuffer;
- private final ByteBuffer readBuffer; // Pre-allocated direct buffer for
zero-copy I/O
+ private final ByteBuffer readBuffer; // Reused per-connection direct
buffer for channel reads (confined to the read thread)
private final Lock readLock = new ReentrantLock();
private final Lock writeLock = new ReentrantLock();
- private volatile boolean open = true;
+ private final AtomicBoolean open = new AtomicBoolean(true);
// Async support
- private final Selector selector;
private volatile Runnable dataListener;
private volatile Consumer<Throwable> disconnectListener;
- private final Thread selectorThread;
+ private final Thread readThread;
public TcpTransportInstance(InetSocketAddress remoteAddress,
TcpTransportConfiguration configuration, AuditLog auditLog) throws
TransportException {
super(configuration, auditLog);
LOGGER.debug("TcpTransportInstance");
this.ringBuffer = new RingBuffer(configuration.receiveBufferSize);
- this.readBuffer = ByteBuffer.allocateDirect(DEFAULT_BUFFER_SIZE); //
Direct buffer for zero-copy
+ this.readBuffer = ByteBuffer.allocateDirect(DEFAULT_BUFFER_SIZE); //
Reused direct buffer for channel reads
try {
// Open socket channel
@@ -90,24 +95,17 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
if (configuration.receiveBufferSize > 0) {
socketChannel.socket().setReceiveBufferSize(configuration.receiveBufferSize);
}
- if (configuration.readTimeout > 0) {
- socketChannel.socket().setSoTimeout(configuration.readTimeout);
- }
+ // Note: configuration.readTimeout is intentionally NOT mapped to
Socket.setSoTimeout here.
+ // SO_TIMEOUT has no effect on blocking SocketChannel reads, so
the previous call was a
+ // silent no-op. Read timeouts are enforced at the protocol/driver
layer (e.g. the S7 driver
+ // bounds responses via CompletableFuture timeouts), not by the
transport.
// Connect with timeout
socketChannel.socket().connect(remoteAddress,
configuration.connectTimeout);
- // Configure non-blocking mode for NIO selector
- socketChannel.configureBlocking(false);
-
- // Create a selector for async I/O
- this.selector = Selector.open();
- socketChannel.register(selector, SelectionKey.OP_READ);
-
- // Start selector thread using virtual thread (Java 21+)
- this.selectorThread = Thread.ofVirtual()
- .name("TCP-Selector-" + remoteAddress.getHostName() + ":" +
remoteAddress.getPort())
- .start(this::runSelectorLoop);
+ // Blocking mode: on Java 21 a virtual thread blocked in
read()/write() parks and
+ // releases its carrier, so no selector is needed.
+ socketChannel.configureBlocking(true);
LOGGER.info("Connected to {}:{} with async support",
remoteAddress.getHostName(), remoteAddress.getPort());
@@ -115,12 +113,19 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
"Connected to: %s:%d with local address: %s:%d",
remoteAddress.getHostName(), remoteAddress.getPort(),
getLocalAddress().getHostName(), getLocalAddress().getPort()));
+
+ // Start the per-connection read loop on a virtual thread (Java
21+) LAST, so an
+ // unchecked throw from the logging/audit above cannot leak an
already-running thread
+ // (the catch only handles IOException and does not stop the read
loop).
+ this.readThread = Thread.ofVirtual()
+ .name("TCP-Read-" + remoteAddress.getHostName() + ":" +
remoteAddress.getPort())
+ .start(this::runReadLoop);
} catch (IOException e) {
String errorMsg = String.format("Failed to connect to %s:%d - %s",
remoteAddress.getHostName(), remoteAddress.getPort(),
e.getMessage());
LOGGER.error(errorMsg, e);
+ // errorMsg already embeds e.getMessage(); a single audit event
avoids a duplicate.
auditLog.write(AuditLogEventType.ERROR, "Error in constructor: " +
errorMsg);
- auditLog.write(AuditLogEventType.ERROR, "Error in constructor: " +
e.getMessage());
throw new TransportException(errorMsg, e);
}
}
@@ -135,7 +140,7 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
@Override
public boolean isOpen() {
- return open && socketChannel.isConnected();
+ return open.get() && socketChannel.isConnected();
}
@Override
@@ -197,10 +202,6 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
// Read and consume bytes
byte[] bytes = ringBuffer.read(numBytes);
- // Re-enable read operations if they were disabled due to full
buffer
- // Now that we've freed up space, the selector can read more data
- reEnableReadIfNeeded();
-
// Log the bytes to the audit log
if (getAuditLog().isEnabled()) {
getAuditLog().write(AuditLogEventType.INCOMING_BYTES,
StaticHelper.ENCODE_HEX(bytes));
@@ -227,44 +228,28 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
ByteBuffer writeBuffer = ByteBuffer.wrap(bytes);
+ // Blocking write parks the virtual thread until the kernel send
buffer accepts the
+ // bytes — natural backpressure, no OP_WRITE registration or sleep
loop needed.
+ // (write() never returns -1; a broken/closed connection surfaces
as
+ // IOException/AsynchronousCloseException, both handled below.)
while (writeBuffer.hasRemaining()) {
- int written = socketChannel.write(writeBuffer);
- if (written == -1) {
- open = false;
- throw new TransportException("Connection closed while
writing");
- }
-
- // If no bytes were written and buffer is full, register for
write operations
- // and wait until the channel becomes writable (prevents CPU
spinning)
- if (written == 0 && writeBuffer.hasRemaining()) {
- try {
- // Temporarily register interest in write operations
- SelectionKey key = socketChannel.keyFor(selector);
- if (key != null && key.isValid()) {
- key.interestOps(key.interestOps() |
SelectionKey.OP_WRITE);
- selector.wakeup();
-
- // Wait a short time for the socket to become
writable
- // This prevents tight CPU spinning when the
socket buffer is full
- Thread.sleep(1);
-
- // Remove write interest to avoid unnecessary
wake-ups
- key.interestOps(SelectionKey.OP_READ);
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new TransportException("Write interrupted", e);
- }
- }
+ socketChannel.write(writeBuffer);
}
- LOGGER.trace("Wrote {} bytes to {}", bytes.length,
socketChannel.getRemoteAddress());
+ LOGGER.trace("Wrote {} bytes", bytes.length);
// Log the bytes to the audit log
if (getAuditLog().isEnabled()) {
getAuditLog().write(AuditLogEventType.OUTGOING_BYTES, "Write:
" + StaticHelper.ENCODE_HEX(bytes));
}
- } catch (TransportException | IOException e) {
+ } catch (AsynchronousCloseException e) {
+ // A concurrent close() closed the channel while we were parked in
write(): normal shutdown.
+ if (!open.get()) {
+ return;
+ }
+ getAuditLog().write(AuditLogEventType.ERROR, "Error in write: " +
e.getMessage());
+ throw new TransportException("Failed to write data", e);
+ } catch (IOException e) {
getAuditLog().write(AuditLogEventType.ERROR, "Error in write: " +
e.getMessage());
throw new TransportException("Failed to write data", e);
} finally {
@@ -274,44 +259,34 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
@Override
public void close() throws TransportException {
- if (!open) {
+ // CAS so concurrent/repeated close() calls run the shutdown exactly
once.
+ if (!open.compareAndSet(true, false)) {
return;
}
- writeLock.lock();
+ // Intentionally takes NO locks: closing the channel is what unblocks
a parked read()/write().
+ // Acquiring writeLock first would deadlock against a writer parked in
a blocking write().
try {
- readLock.lock();
- try {
- open = false;
-
- // Wake up selector
- selector.wakeup();
-
- // Close socket channel
- socketChannel.close();
-
- // Close selector
- selector.close();
-
- // Wait for the selector thread to finish
- if (selectorThread != null) {
- try {
- selectorThread.join(1000);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
+ socketChannel.close();
+ // Only log/audit a successful close here; on failure the catch
reports ERROR
+ // and rethrows, so emitting CLOSE in finally would falsely signal
a clean close.
+ LOGGER.debug("TCP connection closed");
+ getAuditLog().write(AuditLogEventType.CLOSE, "Closed");
+ } catch (IOException e) {
+ getAuditLog().write(AuditLogEventType.ERROR, "Error in close: " +
e.getMessage());
+ throw new TransportException("Failed to close connection", e);
+ } finally {
+ // Always join the read loop, regardless of whether the channel
closed cleanly.
+ // Skip the self-join when close() runs on the read thread itself
(e.g. a
+ // disconnect/data listener calls close()) — joining yourself only
stalls for the
+ // timeout and the loop already exits once open is false.
+ if (readThread != null && Thread.currentThread() != readThread) {
+ try {
+ readThread.join(1000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
}
-
- LOGGER.debug("TCP connection closed");
- getAuditLog().write(AuditLogEventType.CLOSE, "Closed");
- } catch (IOException e) {
- getAuditLog().write(AuditLogEventType.ERROR, "Error in close:
" + e.getMessage());
- throw new TransportException("Failed to close connection", e);
- } finally {
- readLock.unlock();
}
- } finally {
- writeLock.unlock();
}
}
@@ -324,30 +299,6 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
}
}
- /**
- * Re-enables read operations on the selector if they were previously
disabled
- * due to a full ring buffer. This should be called after reading from the
ring
- * buffer to allow new data to be received.
- */
- private void reEnableReadIfNeeded() {
- try {
- SelectionKey key = socketChannel.keyFor(selector);
- if (key != null && key.isValid()) {
- // Check if read interest is currently disabled
- if ((key.interestOps() & SelectionKey.OP_READ) == 0) {
- // Re-enable read operations
- key.interestOps(key.interestOps() | SelectionKey.OP_READ);
- // Wake up the selector to process the new interest ops
- selector.wakeup();
- LOGGER.debug("Re-enabled read operations after buffer
space freed");
- }
- }
- } catch (Exception e) {
- // Log but don't throw - this is a best-effort operation
- LOGGER.warn("Failed to re-enable read operations", e);
- }
- }
-
// ========== AsyncTransportInstance Implementation ==========
@Override
@@ -391,104 +342,76 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
}
/**
- * Selector loop that runs in a virtual thread and notifies listeners when
data arrives.
- * This is the core of the async implementation - no polling needed in the
driver!
+ * Runs the listener on the read thread, guarding against a misbehaving
listener so a thrown
+ * exception cannot silently kill the read loop (mirrors {@link
#notifyDisconnect}'s posture).
*/
- private void runSelectorLoop() {
- LOGGER.debug("Selector loop started");
-
- while (open && !Thread.currentThread().isInterrupted()) {
- try {
- // Block until events are available (no CPU waste!)
- int readyChannels = selector.select();
+ private void safeRun(Runnable listener) {
+ if (listener == null) {
+ return;
+ }
+ try {
+ listener.run();
+ } catch (Throwable t) {
+ LOGGER.error("Data listener failed", t);
+ }
+ }
- if (readyChannels == 0) {
+ /**
+ * Per-connection read loop on a virtual thread: blocking read into the
ring buffer, then
+ * notify the data listener. No selector, no polling.
+ */
+ private void runReadLoop() {
+ LOGGER.debug("Read loop started");
+ try {
+ while (open.get()) {
+ int free = ringBuffer.remainingForWriting();
+ if (free == 0) {
+ // Backpressure: the ring buffer is full and the consumer
has not drained yet.
+ // Park briefly and re-check; never disconnect (only the
codec knows frame
+ // boundaries, and cross-thread consumers like COTP can
legitimately lag).
+ LockSupport.parkNanos(200_000L);
continue;
}
- var selectedKeys = selector.selectedKeys();
- var iterator = selectedKeys.iterator();
-
- while (iterator.hasNext()) {
- SelectionKey key = iterator.next();
- iterator.remove();
-
- if (!key.isValid()) {
- continue;
- }
-
- if (key.isReadable()) {
- // Data available - read it into the ring buffer
- boolean notifyListener = false;
- boolean connectionClosed = false;
- readLock.lock();
- try {
- // Check available space in ring buffer before
reading
- int availableSpace =
ringBuffer.remainingForWriting();
- if (availableSpace == 0) {
- LOGGER.warn("Ring buffer is full, temporarily
disabling read operations");
- key.interestOps(key.interestOps() &
~SelectionKey.OP_READ);
- continue;
- }
-
- // Limit read buffer to available space in
ring-buffer to prevent data loss
- readBuffer.clear();
- readBuffer.limit(Math.min(readBuffer.capacity(),
availableSpace));
-
- int bytesRead = socketChannel.read(readBuffer);
-
- if (bytesRead > 0) {
- readBuffer.flip();
-
- // Write directly from ByteBuffer to ring
buffer (avoiding intermediate byte array allocation)
- int bytesWritten =
ringBuffer.write(readBuffer);
- if (bytesWritten < bytesRead) {
- String message = String.format("Ring
buffer write incomplete. Expected to write " +
- "%d bytes but only wrote %d bytes.
This should not happen.",
- bytesRead, bytesWritten);
- LOGGER.error(message);
-
getAuditLog().write(AuditLogEventType.ERROR, message);
- }
-
- notifyListener = true;
- } else if (bytesRead == -1) {
- // Connection closed gracefully by remote
- LOGGER.info("Connection closed by remote");
- open = false;
- connectionClosed = true;
- }
- } finally {
- readLock.unlock();
- }
-
- // Notify listener OUTSIDE the readLock — the data is
already
- // in the ring buffer, so holding the lock during
response
- // processing would unnecessarily block the I/O path
and cause
- // reentrant lock overhead in processIncomingData().
- if (notifyListener) {
- Runnable listener = dataListener;
- if (listener != null) {
- listener.run();
- }
- } else if (connectionClosed) {
- notifyDisconnect(null);
- break;
- }
- }
+ // Bound the read to free ring-buffer space so the buffer can
never overflow.
+ readBuffer.clear();
+ readBuffer.limit(Math.min(readBuffer.capacity(), free));
+
+ int bytesRead = socketChannel.read(readBuffer); // parks
vthread; releases carrier (JDK21)
+ if (bytesRead == -1) {
+ // Connection closed gracefully by remote
+ LOGGER.info("Connection closed by remote");
+ open.set(false);
+ notifyDisconnect(null);
+ break;
+ }
+ if (bytesRead == 0) {
+ // A blocking read effectively never returns 0; harmless
guard.
+ continue;
}
- } catch (IOException e) {
- getAuditLog().write(AuditLogEventType.ERROR, "Error in
runSelectorLoop: " + e.getMessage());
- if (open) {
- LOGGER.error("Error in selector loop", e);
- open = false;
- notifyDisconnect(e);
+ readBuffer.flip();
+ readLock.lock();
+ try {
+ ringBuffer.write(readBuffer);
+ } finally {
+ readLock.unlock();
}
- break;
+
+ // Notify OUTSIDE readLock — the data is already in the ring
buffer.
+ safeRun(dataListener);
+ }
+ } catch (IOException e) {
+ // If open==false, an intentional close() closed the channel
(AsynchronousCloseException) —
+ // a normal shutdown, not a disconnect. Only a failure while still
open is a real disconnect.
+ if (open.get()) {
+ getAuditLog().write(AuditLogEventType.ERROR, "Error in
runReadLoop: " + e.getMessage());
+ LOGGER.error("Error in read loop", e);
+ open.set(false);
+ notifyDisconnect(e);
}
}
-
- LOGGER.debug("Selector loop stopped");
+ LOGGER.debug("Read loop stopped");
}
}
diff --git
a/plc4j/transports/tcp/src/test/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstanceScalingTest.java
b/plc4j/transports/tcp/src/test/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstanceScalingTest.java
new file mode 100644
index 0000000000..4c994abb5d
--- /dev/null
+++
b/plc4j/transports/tcp/src/test/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstanceScalingTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.plc4x.java.transport.tcp;
+
+import org.apache.plc4x.java.transport.tcp.config.TcpTransportConfiguration;
+import org.apache.plc4x.java.utils.auditlog.api.AuditLog;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+import java.net.InetSocketAddress;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Concurrency-model probe: opens many idle connections and reports how many
carrier (platform)
+ * threads the virtual-thread scheduler is using. With the old selector model,
each connection's
+ * virtual thread blocks in {@code Selector.select()}, which does not unmount
the virtual thread,
+ * so the scheduler ties up a carrier per connection (toward maxPoolSize=256)
— measured at 201
+ * carriers for 200 connections on both JDK 21 and JDK 25, i.e. NOT the
synchronized-monitor
+ * pinning that JEP 491 (JDK 24) removed. With the blocking-read model, the
read virtual threads
+ * unmount via the NIO poller, so the carrier pool stays small and flat
regardless of connection
+ * count (measured 2-3 for 200 connections).
+ *
+ * The test asserts only that all connections stay open (lenient — it is an
evidence probe, not a
+ * brittle threshold). The before/after signal is the {@code
CARRIER_COUNT=...} line printed to
+ * stdout and, when run with {@code -Djdk.tracePinnedThreads=full}, the
presence/absence of
+ * pinned-thread stack traces.
+ */
+class TcpTransportInstanceScalingTest {
+
+ private static final int CONNECTIONS = 200;
+
+ @Test
+ @Disabled("Concurrency-model evidence probe — opens 200 sockets and
sleeps; run manually, "
+ + "ideally with -Djdk.tracePinnedThreads=full. Not a CI regression
test.")
+ void manyIdleConnections_carrierThreadUsage() throws Exception {
+ ServerSocketChannel server = ServerSocketChannel.open();
+ server.bind(new InetSocketAddress("localhost", 0), 256);
+ int port = ((InetSocketAddress) server.getLocalAddress()).getPort();
+
+ List<SocketChannel> serverSide = new ArrayList<>();
+ volatileFlag.running = true;
+ Thread acceptThread = new Thread(() -> {
+ try {
+ while (volatileFlag.running) {
+ SocketChannel s = server.accept();
+ synchronized (serverSide) {
+ serverSide.add(s);
+ }
+ }
+ } catch (Exception ignored) {
+ // server closed
+ }
+ }, "scaling-accept");
+ acceptThread.setDaemon(true);
+ acceptThread.start();
+
+ List<TcpTransportInstance> clients = new ArrayList<>();
+ try {
+ for (int i = 0; i < CONNECTIONS; i++) {
+ TcpTransportConfiguration config = new
TcpTransportConfiguration();
+ config.receiveBufferSize = 81920;
+ config.connectTimeout = 5000;
+ clients.add(new TcpTransportInstance(
+ new InetSocketAddress("localhost", port), config,
AuditLog.builder().build()));
+ }
+
+ // Let every connection's read virtual thread settle into its
blocking wait.
+ Thread.sleep(3000);
+
+ // One snapshot so carriers and total are derived from the same
instant.
+ Set<Thread> threads = Thread.getAllStackTraces().keySet();
+ // Count only virtual-thread scheduler carriers, not
ForkJoinPool.commonPool
+ // workers (parallel streams etc.), which would inflate the
carrier count.
+ long carriers = threads.stream()
+ .filter(t -> !t.isVirtual())
+ .filter(t -> t.getName().contains("ForkJoinPool"))
+ .filter(t -> !t.getName().contains("commonPool"))
+ .count();
+ long total = threads.size();
+ System.out.println("CARRIER_COUNT=" + carriers
+ + " TOTAL_THREADS=" + total
+ + " CONNECTIONS=" + CONNECTIONS
+ + " CPUS=" + Runtime.getRuntime().availableProcessors());
+
+ long openCount =
clients.stream().filter(TcpTransportInstance::isOpen).count();
+ assertTrue(openCount >= CONNECTIONS - 5,
+ "expected ~all connections open, got " + openCount + "/" +
CONNECTIONS);
+ } finally {
+ for (TcpTransportInstance c : clients) {
+ try { c.close(); } catch (Exception ignored) { }
+ }
+ volatileFlag.running = false;
+ synchronized (serverSide) {
+ for (SocketChannel s : serverSide) {
+ try { s.close(); } catch (Exception ignored) { }
+ }
+ }
+ server.close();
+ }
+ }
+
+ // tiny holder carrying a volatile stop flag the daemon accept loop polls
to shut down
+ private static final class Flag { volatile boolean running; }
+ private final Flag volatileFlag = new Flag();
+}
