From bca46bd89d2dcbd21f9aa295b17fea81a4a4907d Mon Sep 17 00:00:00 2001 From: Jooyoung Jung Date: Fri, 19 Jun 2026 09:34:27 +0900 Subject: [PATCH 1/3] 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. --- .../transport/tcp/TcpTransportInstance.java | 311 +++++++----------- .../tcp/TcpTransportInstanceScalingTest.java | 121 +++++++ 2 files changed, 235 insertions(+), 197 deletions(-) create mode 100644 plc4j/transports/tcp/src/test/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstanceScalingTest.java 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 96e2df38ccc..511d8464e4e 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.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 implements AsyncTransportInstance { @@ -52,22 +58,21 @@ public class TcpTransportInstance extends BaseTransportInstance 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,22 @@ public TcpTransportInstance(InetSocketAddress remoteAddress, TcpTransportConfigu 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); + // 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); - // Start selector thread using virtual thread (Java 21+) - this.selectorThread = Thread.ofVirtual() - .name("TCP-Selector-" + remoteAddress.getHostName() + ":" + remoteAddress.getPort()) - .start(this::runSelectorLoop); + // Start the per-connection read loop on a virtual thread (Java 21+) + this.readThread = Thread.ofVirtual() + .name("TCP-Read-" + remoteAddress.getHostName() + ":" + remoteAddress.getPort()) + .start(this::runReadLoop); LOGGER.info("Connected to {}:{} with async support", remoteAddress.getHostName(), remoteAddress.getPort()); @@ -135,7 +138,7 @@ public InetSocketAddress getLocalAddress() { @Override public boolean isOpen() { - return open && socketChannel.isConnected(); + return open.get() && socketChannel.isConnected(); } @Override @@ -197,10 +200,6 @@ public byte[] read(int numBytes) throws TransportException { // 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 +226,30 @@ public void write(byte[] bytes) throws TransportException { 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. while (writeBuffer.hasRemaining()) { int written = socketChannel.write(writeBuffer); if (written == -1) { - open = false; + open.set(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); - } - } } - 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,28 @@ public void write(byte[] bytes) throws TransportException { @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(); + } catch (IOException e) { + getAuditLog().write(AuditLogEventType.ERROR, "Error in close: " + e.getMessage()); + throw new TransportException("Failed to close connection", e); + } finally { + if (readThread != null) { + 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(); + LOGGER.debug("TCP connection closed"); + getAuditLog().write(AuditLogEventType.CLOSE, "Closed"); } } @@ -324,30 +293,6 @@ private void ensureOpen() throws TransportException { } } - /** - * 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 +336,76 @@ private void notifyDisconnect(Throwable cause) { } /** - * 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 00000000000..0b9ffee6669 --- /dev/null +++ b/plc4j/transports/tcp/src/test/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstanceScalingTest.java @@ -0,0 +1,121 @@ +/* + * 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 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 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 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); + + long carriers = Thread.getAllStackTraces().keySet().stream() + .filter(t -> !t.isVirtual()) + .filter(t -> t.getName().contains("ForkJoinPool")) + .count(); + long total = Thread.getAllStackTraces().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 so the daemon accept loop can observe a stop flag without a field on the test + private static final class Flag { volatile boolean running; } + private final Flag volatileFlag = new Flag(); +} From 469e1f53263853058f2636f9f31beea3965e3800 Mon Sep 17 00:00:00 2001 From: Jooyoung Jung Date: Wed, 24 Jun 2026 02:04:55 +0900 Subject: [PATCH 2/3] 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). --- .../plc4x/java/transport/tcp/TcpTransportInstance.java | 7 +++++-- .../transport/tcp/TcpTransportInstanceScalingTest.java | 2 +- 2 files changed, 6 insertions(+), 3 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 511d8464e4e..52db9050bda 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 @@ -268,10 +268,15 @@ public void close() throws TransportException { // Acquiring writeLock first would deadlock against a writer parked in a blocking write(). try { 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. if (readThread != null) { try { readThread.join(1000); @@ -279,8 +284,6 @@ public void close() throws TransportException { Thread.currentThread().interrupt(); } } - LOGGER.debug("TCP connection closed"); - getAuditLog().write(AuditLogEventType.CLOSE, "Closed"); } } 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 index 0b9ffee6669..c4d381bd2f0 100644 --- 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 @@ -115,7 +115,7 @@ void manyIdleConnections_carrierThreadUsage() throws Exception { } } - // tiny holder so the daemon accept loop can observe a stop flag without a field on the test + // 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(); } From 0a1e480ddb1eb35b07c7b84fba80cefb52d54f55 Mon Sep 17 00:00:00 2001 From: Jooyoung Jung Date: Wed, 24 Jun 2026 02:38:15 +0900 Subject: [PATCH 3/3] fix(transport-tcp): drop dead write -1 check and tidy audit/test nits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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. --- .../transport/tcp/TcpTransportInstance.java | 27 ++++++++++--------- .../tcp/TcpTransportInstanceScalingTest.java | 10 +++++-- 2 files changed, 23 insertions(+), 14 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 52db9050bda..6b67613c059 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 @@ -107,23 +107,25 @@ public TcpTransportInstance(InetSocketAddress remoteAddress, TcpTransportConfigu // releases its carrier, so no selector is needed. socketChannel.configureBlocking(true); - // Start the per-connection read loop on a virtual thread (Java 21+) - this.readThread = Thread.ofVirtual() - .name("TCP-Read-" + remoteAddress.getHostName() + ":" + remoteAddress.getPort()) - .start(this::runReadLoop); - LOGGER.info("Connected to {}:{} with async support", remoteAddress.getHostName(), remoteAddress.getPort()); auditLog.write(AuditLogEventType.CONNECT, String.format( "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); } } @@ -228,12 +230,10 @@ public void write(byte[] bytes) throws TransportException { // 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.set(false); - throw new TransportException("Connection closed while writing"); - } + socketChannel.write(writeBuffer); } LOGGER.trace("Wrote {} bytes", bytes.length); @@ -277,7 +277,10 @@ public void close() throws TransportException { throw new TransportException("Failed to close connection", e); } finally { // Always join the read loop, regardless of whether the channel closed cleanly. - if (readThread != null) { + // 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) { 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 index c4d381bd2f0..4c994abb5de 100644 --- 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 @@ -28,6 +28,7 @@ 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; @@ -88,11 +89,16 @@ void manyIdleConnections_carrierThreadUsage() throws Exception { // Let every connection's read virtual thread settle into its blocking wait. Thread.sleep(3000); - long carriers = Thread.getAllStackTraces().keySet().stream() + // One snapshot so carriers and total are derived from the same instant. + Set 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 = Thread.getAllStackTraces().size(); + long total = threads.size(); System.out.println("CARRIER_COUNT=" + carriers + " TOTAL_THREADS=" + total + " CONNECTIONS=" + CONNECTIONS