LivingLikeKrillin commented on code in PR #2612:
URL: https://github.com/apache/plc4x/pull/2612#discussion_r3461617459
##########
plc4j/transports/tcp/src/main/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstance.java:
##########
@@ -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");
Review Comment:
Fixed in 469e1f53263853058f2636f9f31beea3965e3800. Moved the "TCP connection
closed" debug line and the CLOSE audit event into the success path so they only
fire when socketChannel.close() succeeds; on failure the catch reports the
ERROR event and rethrows. readThread.join() stays in finally so the read loop
is always awaited.
##########
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<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);
+
+ 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();
Review Comment:
Fixed in 469e1f53263853058f2636f9f31beea3965e3800. Reworded the comment to
describe the holder accurately (it carries a volatile stop flag the daemon
accept loop polls); the previous "without a field" wording was inaccurate since
the holder is itself a field.
--
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]
